Processing Pipelines with TPL Dataflow

Pipelining is a very well known design pattern. It is used when a stream of data elements are processed through a series of pre-determined steps where a output of one serves as input for other step. These steps might also be conditionally linked. The concept is similar to assembly line used in factories where a stream of raw material is continuously added which goes through a series of steps to a finished product. This is commonly used in manufacturing plants. It must be remembered that pipelining has no effect on the processing of a single element (responsiveness) but it is phenomenal to improve the throughput of the overall system.

Let's consider a simple dataflow pipeline where a data element is passed through square, offset and doubling stages. The output of each step is used as input of consecutive following step. A list of element needs to be passed through these stages which would result in a resulting output list with processed elements.



In order to abstract the implementation of the processor, let us introduce an implementation. We will be providing different implementation of the processor and discuss the details about them.

Linear Processor
This is the simplest implementation of the processing pipeline. This divides the processing steps in different methods. We apply the same series of operations on all elements in loop. We also keep adding the resulting elements in a separate collection. When the loop finishes execution, we have our desired results in the collection, which is the returned.

Although this is the simplest and easiest of all data flow pipeline implementation but it doesn't make use of any parallelism in the code. The time to execute all elements in the pipeline is linearly proportional to the size of input collection. So if we assign x, y and z time units to execute square, offset and double operations than to process n elements it would require n(x + y + z) time units.

For more complex steps, we might want to refactor our code by extracting the implementation of these operations into separate type. This would make unit testing these steps a lot easier.

Parallel Loop Processor
Based on the recommendations from Pattern & Practices about usage of Parallel Loop pattern, it is suitable when we need to apply independent operations to a list of elements. This fits to our requirement where we need to apply a series of independent operations to a list of elements. These independent operations are Square, Offset and Double. For simplicity sake, we are not discussing cancellation and exception handling for these loops. Here we have also made use of System.Collections.Concurrent.ConcurrentBag<T>. This type was introduced in .net framework 4.0. This is one of the implementation of IProducerConsumerCollection<T> also introduced in the same framework version. ConcurrentBag has introduced a shared state but this is a thread safe type.

The only caveat is actually a feature of Parallel loop pattern. We cannot determine the actual order of execution of elements. We could verify that by looking at the returned data from Process() method. In this case, it seems to be a requirement that order of result should be the same as the input collection. So the result from this processor would be logically wrong.

Pipeline Processor
As we discussed above, we cannot use Parallel loop pattern when the order of elements in the result matters. Pattern & Practice team recommends another pattern for such requirement. This pattern is called Pipelining. The pattern is used in situations where the elements of a collection passes through a series of steps where output of one step serves as input to the next step. This pattern improves the throughput of multi-core machines by distributing the steps to multiple cores.



Pipelining requires us to update the processing elements to use producer consumer based collections. Here each step produces elements which are then consumed by the next processing element in the pipeline. Producer / Consumer based collections in the processing pipeline make each step independent of immediate previous and next element. The time to execute a list of operations through a pipeline is over-shadowed by the slowest element in the processing pipeline.

In the following example, we have used an implementation of IProducerConsumerCollection<T> i.e. BlockingCollection<T>. This was also introduced in .net framework 4.0. This provides an ideal implementation for producer consumer scenarios. Here consumers can wait using GetConsumingEnumerable() method of the collection. As the elements are added to the collection, they are enumerated by consumers. This also supports notifying the consumers about the completion of the producer / consumer scenario. In this case, no more element can be added to the collection. After enumerating the existing elements in the collection, the enumeration ends. It is a very common bug to forget calling GetConsumingEnumerable methods and directly iterating through the collection. Even I made the same mistake when I was writing the below code. In that case, the loop finishes if there are no current element in the collection.

As we finish processing all elements of input collection, we can notify the consumers by calling CompleteAdding method of BlockingCollection<T>. The consumers waiting on the collection finish execution of the elements in the collection and complete iterating the collection. In the case of processing pipeline, we need to call the same operation on the output collection for the processing element as the next consecutive processing element is waiting for elements in this collection. In this way, the elements in the processing pipeline pass the signal for execution completion. We can return the resulting collection to the requester now. We need to wait until the last element in the processing pipeline finishes execution. Since all the processing elements are implemented through task we can just call Wait on the last task in the pipeline to wait for finishing the execution of processing pipeline.

Dataflow Block Processor
Microsoft introduced Tpl Data flow to make it easier to construct data flow pipelines and networks. This hasn't been released as part of .net framework yet but you can get it as a nuget package.



After installing the package you should see the following reference being added to assembly references section.



There are three main interfaces in TPL Dataflow library. They are ISourceBlock, ITargetBlock and IPropagatorBlock. All blocks are inherited from IDataflowBlock. This becomes very useful while chaining of different blocks to construct a processing pipeline. Any source block can be linked to another Data flow block. In case of multiple destination blocks, it can route message to whoever picks it first. This is common to producer / consumer scenario. It also supports Broadcast block which could route message to multiple destination blocks.



From the coding perspective, Dataflow blocks based approach seems to be the middle ground between Linear Process and Pipeline processors discussed above. We don't need to create separate tasks manually and use Blocking collection, so it is nearly as simple as linear processor. It supports producer / consumer scenario by incorporating internal input and output queues which makes pipelining possible. We can implement our dataflow pipeline as in the following diagram:



We need a combination block to keep adding the results in a collection. When the processing pipeline finishes execution, it just returns the consolidated results. Source and Propagation blocks are a lot like observable collections where the next element is passed to the recipient without the recipients needing to ask for this.

In the above code, we are constructing the whole pipeline in the constructor of the processor. If the data flow blocks are pure in nature, we can even do that in a static constructor keeping the blocks as class members. As depicted in the previous image, we have created and linked four blocks to square, offset, double and consolidate the results.

We also need to chain the completion signal. As the first block receives the completion signal, the message is propagated in the data flow pipeline. Calling CompleteAdding() on a dataflow block means that there are no more blocks expected. After finishing executing the already existing elements in the queue, it can pass the message to the next block in the pipeline (if configured). At the end, we are waiting so that the last element in the pipeline finishes processing by using IDataflowBlock.Completion . We are just returning the results after that.

Dataflow Blocks also support handling exceptions and cancellation during the block execution. Let's keep that for a later discussion.

Download

Microsoft Accelerator Research Project

As you know we can great performance benefits from the new .net 4.0 features like Task Parallel Library (TPL), Coordination Data Structures (CDS) and Parallel LINQ (PLINQ). But all of these features revolve around Task Parallelism. Now what about data parallelism? Should I be writing a Parellel.ForEach just to add elements of an integer array?

To provide a tool, for this need, in the toolkit of developer, there has been work going on in Microsoft for sometime. This is called Microsoft Accelerator Project. The whole idea around Accelerator is to provide implicit parallelism for array processing through functional constructs provided through managed wrapper in any .net language. We must consider that this is still a research project. Microsoft might or might not release it ever. But all we want to see is about the features that this toolkit exposes.

In this article I would not be discussing about the detailed architecture of Acclerator. I would just be presenting the introduction of this framework so that an application developer might be using it easily using the functional programming constructs provided in the toolkit. In parallel processing world, a vector is a one-dimensional array. We have always been in need to exploit parallel operations in our software. Using accelerator, you would really have the following resulting logical behavior.

var arr3 = arr1 + arr2;

Here arr1 and arr2 may be two numeric arrays. This is evident from above statement that Accelerator is based on SIMD (Single Instruction Multiple Data) architecture of Flynn’s taxonomy.

Target Platforms:

When Accelerator was first introduced, the whole idea was to use the capabilities (large numbers of arithmetic units) of graphics processor for computations even in non-graphic applications to gain huge speed-ups. But after the introduction of multicore machines in general use, it was updated to take advantage of multiple cores. Now the developer has option to use either of graphics processor (GPU) or multiple main cores in the machine.

With Acclerator, we can target the code generation for two platforms. They are as follows:

1. DX9: To run the computation on GPU. When we specify this target then the code is JIT compiled into pixel shader code and pushed to GPU for processing. After computation, the result is pushed back to the CPU and the result is available for the main program.

2. X64MulticoreTarget: To run the computation on a multicore CPU (only 64 bits).

How it works?

Accelerator uses delayed processing (Lazy computations) to perform its computations. It keeps on adding all the operations in the form of a graph. As the framework receives instructions to execute the computations, it translates the graph based on the platform selected (DX9 or X64MulticoreTarget). It sends the generated code the target and receives result where it gets executed. These results are then provided to the calling program.

This is the same process as in LINQ. It maintains the operations and operands in the form of Directed Acyclic Graphs (DAG) in order to execute them later on. This DAG is called Expression Graph. After computation the Data Parallel object is converted to System.Object object (Array or Bitmap).

The idea is load your arrays in DataParallelArray objects (discussed later). Perform computations on them. Convert them to System.Object back when the computation is done.

Accelerator Architecture:

ParallelArray:

The architecture of accelerator is based on ParallelArray class. This is the base class of 4 classes available. They are IntParallelArray, FloatParallelArray, Float4ParallelArray and BoolParallelArray. They are further specialized with their disposable versions like DisposableIntParallelArray, DisposableFloatParallelArray, DisposableFloat4ParallelArray and DisposableBoolParallelArray.

ParallelArrays:

It is a set of static methods for parallel operations. These operations also include initialization and uninitialization CPU and GPU accelerator. Some of these operations are also available through the overloaded operators for array by Accelerator.

Exception:

The framework also supports exceptions. Currently only UnexpectedOperation exception is supported.

Using Accelerator in .net Applications:

To use accelerator in .net projects, following steps should be followed:

• Install Accelerator. The latest download (msi) can be obtained from the following Microsoft Connect page. https://connect.microsoft.com/acceleratorv2

• Add a reference of Microsoft.Accelerator.dll in your project. The library should be available in the installation directory of Microsoft Accelerator.

• Add Microsoft.ParallelArrays namespace to the desired namespace. It includes all the accelerator types to write data parallel operations.

• Copy Accelerator.dll from the installation folder of Microsoft Acclerator v2 (…\Microsoft\Accelerator v2\bin\x86\Release) to the application folder. Since I am running in on an x86 machine, I have copied x86 version of the dll. You might use x64 one for a 64-bit machine.

Operations supported:

• Element Wise operations

In this kind of operation, an operation is applied between corresponding elements of an array. The dimensions of the operands should be same for this kind of operations. In the following example, we have added the elements of two float arrays. The results are available in result float array. As expected, the result is {2.0f, 4.0f, 6.0f, 8.0f, 10.0f}.

float[] arr1 = { 1.0f, 2.0f, 3.0f, 4.0f, 5.0f };

float[] arr2 = { 1.0f, 2.0f, 3.0f, 4.0f, 5.0f };

var acc_arr1 = new FloatParallelArray(arr1);

var acc_arr2 = new FloatParallelArray(arr2);

var acc_arr3 = ParallelArrays.Add(acc_arr1, acc_arr2);

var target = new DX9Target();

var result = target.ToArray1D(acc_arr3);

Since ‘+’ operator is overloaded, we could write the same operation as follows:

var acc_arr3 = acc_arr1 + acc_arr2;

• Reductions (sum, product, min, max etc).

Accelerator supports reduction operations using ParallelArrays. In the following example, we are obtaining sum of all elements of a float array.

float[] arr1 = { 1.0f, 2.0f, 3.0f, 4.0f, 5.0f };

var acc_arr1 = new FloatParallelArray(arr1);

var acc_result = ParallelArrays.Sum(acc_arr1);

var target = new DX9Target();

var result = target.ToArray1D(acc_result);

• Conversions between Ints and Floats:

Not only we can do operations on arrays, we can also convert between them. In the following example, we are converting an int array to a float one.

int[] arr1 = { 1, 2, 3, 4 };

var acc_arr1 = new IntParallelArray(arr1);

var acc_result = ParallelArrays.ToFloatParallelArray(acc_arr1);

var target = new DX9Target();

var result = target.ToArray1D(acc_result);

To verify the actual conversion, IDE can also help during debugging.

• Logical operations:

Accelerator may be used for performing various logical operations on operands. These operations might be boolean (Not, AND, OR), comparison (Less than, GreaterThan) and selection operations (Selection). Currently most of these operations are not implemented completely.

float[] arr1 = { 1.0f, 2.0f, 3.0f, 4.0f };

float[] arr2 = { 0.0f, 1.0f, 7.0f, 3.0f };

bool[] arr3 = { false, true, false, true };

var acc_arr1 = new FloatParallelArray(arr1);

var acc_arr2 = new FloatParallelArray(arr2);

var acc_arr3 = new BoolParallelArray(arr3);

var acc_result = ParallelArrays.Cond(acc_arr3, acc_arr1, acc_arr2);

var target = new DX9Target();

var result = target.ToArray1D(acc_result);

The above code should result in {0.0f, 2.0f, 7.0f, 4.0f} in result. But currently, it has {0.0f, 1.0f, 7.0f, 3.0f} (which is definitely an issue).

• Transformations on data parallel arrays (shift, duplicate):

Some operations require transformation of arrays to be used in an operation. In the following example, we are creating an array. The elements of the result array are the sum of the element in the original array in the specified position and the element before it.

float[] arr1 = { 1, 2, 3, 4 };

var acc_arr1 = new FloatParallelArray(arr1);

var acc_result = ParallelArrays.Add(ParallelArrays.Shift(acc_arr1, -1), acc_arr1);

var target = new DX9Target();

var result = target.ToArray1D(acc_result);

Some operations require transformation of arrays to be used in an operation. In the following example, we are creating an array. The elements of the result array are the sum of the element in the original array in the specified position and the element before it.

This explains that (-1) has shifted each element 1 position to the right keeping the 1^st element and truncating the last one.

• Matrix computations:

There are many matrix operations which are directly available in Acclerator. The other operations can be implemented using already existing element wise, transformation and reduction operations available.

In the following example, we are creating the transpose of a matrix.

float[,] arr1 = {

{ 1, 2, 3, 4 },

{ 1, 2, 3, 4 },

{ 1, 2, 3, 4 },

{ 1, 2, 3, 4 }

};

var acc_arr1 = new FloatParallelArray(arr1);

var acc_result = ParallelArrays.Transpose(acc_arr1);

var target = new DX9Target();

var result = target.ToArray2D(acc_result);

You can see that instead of using ToArray1D method from target, we have used ToArray2D operation. The result array should be as follows:

{

{ 1, 1, 1, 1 },

{ 2, 2, 2, 2 },

{ 3, 3, 3, 3 },

{ 4, 4, 4, 4 }

};

Application Areas:

The main areas which have potential for extensive use of Accelerator are compression and image processing. These are the areas where there is large number of array processing and matrix manipulation operations.