In our previous TPL Dataflow post we introduced the concept of linking TPL Dataflow blocks together in a processing chain. The example we used was effective in illustrating the point. It was a simple, yet not very practical example showing how we can link multiple TBL Dataflow blocks into a network for propagating and processing data to get to an end result.

Let's consider another example that illustrates a more practical scenario where we have data coming in from multiple sources that needs to be combined, transformed and then propagated out to an end point such as user interface. But first we'll need introduce two additional TPL dataflow blocks that will help illustrate these concepts. These two blocks are BufferBlock<T> and JoinBlock<T>.

BufferBlock<T> is a dataflow block for buffering data from a source and propagating it out to a target. It receives incoming data from some source and buffers it up as output to be consumed by some other entity down the propagation chain. JoinBlock<Tin, Tout> accepts two inputs and joins them together into tuples, buffering them as output to be consumed by some other entity in the propagation chain. By themselves, BufferBlock<T> and JoinBlock<Tin, Tout> are not very interesting, but when combined with other TPL Dataflow blocks they can be used to create complex dataflow networks for developing robust messaging based applications that process data in an asynchronous and concurrent fashion.

Let's consider a scenario where we have integer data coming in from two different sources that we need to add together and present to a user input. To accomplish this we'll use two BufferBlock<T> blocks representing the two data sources. We'll combine the inputs from the two data sources into tuples using JoinBlock<Tin, Tout> so that they can be added together further down the chain. We'll use a TransformBlock<Tin, Tout> to add the items in each tuple, and finally we will use ActionBlock<T> will present the sum to the console. The dataflow network for this scenario would look something like the diagram show below.

The code for this is very straight forward. We'll configure each block, link them together according to the dataflow graph pictured above and then post the data to each of the BufferBlock<T> blocks which in turn will be propagated through the data network where it will be transformed and presented to the console. Here's the code

class Program

{

    static void Main(string[] args)

    {

        var buf1 = new BufferBlock<int>();

        var buf2 = new BufferBlock<int>();

        var join = new JoinBlock<int, int>();

        var transform = new TransformBlock<Tuple<int, int>, string>(t =>

        {

            return string.Format("{0} + {1} = {2}", t.Item1, t.Item2, t.Item1 + t.Item2);

        });

        var action = new ActionBlock<string>(s =>

        {

            Thread.Sleep(500);

            Console.WriteLine(s);

        });

        buf1.LinkTo(join.Target1);

        buf2.LinkTo(join.Target2);

        join.LinkTo(transform);

        transform.LinkTo(action);

        for (int i = 0; i < 20; i++)

        {

            if (i % 2 == 0)

                buf1.Post(i);

            else

                buf2.Post(i);

        }

        Console.WriteLine("Posting Complete!");

        Console.ReadLine();

    }

}

Here's the output:

As you can see, the data was posted to the buffer blocks immediately, indicated by the Posting Complete message. Then processing kicked in, are buffered data was combined into tuples, transformed and then presented the result to the console.

In this series of posts we introduced several of the concepts being exposed in the next version of the Visual Studio .NET and the .NET Framework. I personally think these are some pretty cool capabilities that we can look forward to. You can start learning about them today by downloading the Async CTP. The examples in these posts will work with Version 3 of the Async CTP released in October 2011 available at http://www.microsoft.com/download/en/details.aspx?displaylang=en&id=9983, as well as the September 2011 release of Visual Studio 11 Developer preview available at http://www.microsoft.com/download/en/details.aspx?displaylang=en&id=27543.

In our previous posts on TPL Dataflow we talked about ActionBlock<T> and TransformBlock<Tin, Tout>, two dataflow blocks exposed in the TPL Dataflow library that allow you to process message based data asynchronously in a parallel manner. These two blocks, in and of themselves, are quite useful, but there are additional blocks, and additional capabilities available in TPL Dataflow that we can take advantage of that you may want to concern yourself with.

One of the most important aspects of the TPL Dataflow library is the ability to link multiple blocks together. Dataflow blocks can be chained together to form arbitrary dataflow networks where the output from one block can be exposed as inputs to other blocks, or where the output from many different blocks needs to be joined into a single output and exposed to one or more blocks further down in the processing chain. There are many scenarios in which this can be useful, such as a stock hit application where data about stocks is coming in from multiple sources, or when acquiring data from a biometric feedback device where you might be collecting data from multiple channels to analyze and report on different aspects of biometric data from a patient such as blood pressure, heart rate or ECG.

Let's take a look out how we can use TPL Dataflow to solve these types of problems. TPL DataFlow has the ability to link the input of a target block to the output of a source block via the ISourceBlock.LinkTo() method that is available on every source block type. Building on the examples from our previous posts where we described ActionBlock<T> and TransformBlock<Tin, Tout>, we'll take advantage of the ISourceBlock.LinkTo() method to link TransformBlock<Tin, Tout> block's output to ActionBlock<T> block's input.

In this example we'll create an ActionBlock<T> that will accept incoming data, sleep briefly and then output the incoming data to the console. We'll also create a TransformBlock<Tin, Tout> that will accept incoming integer data, transform that data by doubling it and then place the transformed data on the output queue for further processing. Finally we'll link the TransformBlock<Tin, Tout> output to the ActionBlock<T> input using the LinkTo() method of the TransformBlock<Tin, Tout> and then post some data. Here's the code:

class Program

{

    static void Main(string[] args)

    {

        var action = new ActionBlock<int>(i =>

        {

            Thread.Sleep(500);

            Console.WriteLine(i);

        });

        var transform = new TransformBlock<int, int>(i => i * 2);

        transform.LinkTo(action);

        for (int i = 0; i < 10; i++) transform.Post(i);

        Console.WriteLine("Posting Complete!");

        Console.ReadLine();

    }

}

Here's the output:

As you can see, the data was posted to TransformBlock<Tin, Tout>, multiplied by 2, and then propagated to ActionBlock<T> via the ISourceBlock.LinkTo link where it was presented to the console.

This was a simple to illustrate the concept of linking the output of one block to the input of another in a processing chain. In our next post we'll take this a step further by showing you how we can buffer our output and consume that buffered output further down in the processing chain using the BufferBlock<T> block. We'll also take a look at out we can consume data from multiple sources and combine them using JoinBlock<Tin, Tout> for further processing later in the processing chain.

In our previous TPL Dataflow posts we introduced the TPL Dataflow Library and showed you some of the capabilities of ActionBlock<T>, a dataflow block available in TPL Dataflow that allows you to buffer up messages and process those messages concurrently. We also took a slight detour and discussed how we can control the lifetime of our tasks through some of the completion mechanisms available via the IDataflowBlock interface. In this post we'll continue the main Async CTP TPL Dataflow discussion and introduce another TPL Dataflow block called TransformBlock<Tin, Tout>.

TransformBlock<Tin, Tout>, for those familiar with LINQ, is somewhat similar to Select in that it takes an input, transforms that input in some manner, and then produces an output. Like ActionBlock<T>, a TransformBlock<Tin, Tout> buffers all of its input data and processes that input data asynchronously using a function that we provide. By default TransformBlock<Tin, Tout> processes its data sequentially with a MaxDegreeOfParallelism equal to 1. This of course can be controlled using ExecutionDataflowBlockOptions.

The interesting thing to note about TransformBlock<Tin, Tout> is that in addition to receiving buffered input and processing it, TransformBlock<Tin, Tout> will take all of its processed output and buffer that as well. Are you starting to see a theme here?

Let's take a look at an example using TransformBlock<Tin, Tout> that transforms each integer into its double.

class Program

{

    static void Main(string[] args)

    {

        var transform = new TransformBlock<int, int>(i => i * 2);

        for (int i = 0; i < 10; i++) transform.Post(i);

        Console.ReadLine();

    }

Not a very interesting example, really. We provided a Func<> to TransformBlock<Tin, Tout> that returns each input multiplied by 2. As we expect the data is posted to the block and buffered for processing. TransformBlock<Tin, Tout> executes the provided Func<> and the results are then buffered for further processing by some other mechanism in our application. But how do we know that our input and output are being buffered? Let's take advantage of the debugger and find out for sure.

Using the Visual Studio Async CTP create a console app containing the following code and run it under the Visual Studio Debug configuration.

class Program

{

    static void Main(string[] args)

    {

        var transform = new TransformBlock<int, int>(i =>

            {

                Console.WriteLine(i * 2);

                return i * 2;

            });

        for (int i = 0; i < 10; i++) transform.Post(i);

        Debugger.Break();

        Thread.Sleep(1000);//Give the TransformBlock some time to process the input

        Debugger.Break();

        Console.ReadLine();

    }

Notice that the Visual Studio debugger stopped at the Debugger.Break statement. Now hover over the transform instance variable and you'll see that the current InputCount showing that immediately after posting all of the data 10 items have been buffered for processing.

Also note that the OutputCount shows zero items because we broke into the debugger just prior to processing the data.

Now continue executing in Visual Studio until the second Debugger.Break statement and hover your mouse pointer over the transform instance variable again to verify the current state of TransformBlock<Tin, Tout>. You can see that the OutputCount is now equal to 10, indicating that the processing occurred and that 10 items were processed and queued up in the output buffer.

Still not a very interesting example. How do we know that each input was actually doubled and placed into the output queue? So let's make sure that TransformBlock<Tin, Tout> did indeed process our data. Run the example again until the debugger breaks at the second Debugger.Break statement. This time hover over the transform instance variable and drill down into the instance to the OutputQueue, and notice that it indeed contains the double of each of our inputs.

There are variety of capabilities available on these TPL Dataflow blocks that further enhance your ability to process data using parallel patterns. We've only scratched the surface here with the introduction of the TPL Dataflow ActionBlock<T> and TransformBlock<Tin, Tout> along with some of their capabilities. In our next post we'll introduce additional features and capabilities of TPL Dataflow, including the ability to link different dataflow blocks together to create arbitrary dataflow networks that asynchronously propagate data through a series of processing steps to get our data into its desired form.

In our previous posts we introduced the concept of TPL Dataflow and even showed how you can take advantage of ActionBlock<T> to process incoming data in parallel. We also showed how you can control the degree of parallelism applied in a parallel dataflow through the use of the ExecutionDataflowBlockOptions MaxDegreeOfParallelism property. In this post we'll introduce the concept of task completion in a parallel dataflow.

Often times when dealing with these types of data loads in your applications you may want to post a large amount of asynchronous data for processing and then be notified when that processing has completed. We can do this easily in TPL DataFlow because each TPL Dataflow block has a lifetime, and each block type exposes mechanisms that allow us to monitor and control that lifetime such that we can wait for the tasks to complete all of their processing, or be notified when they have completed their processing. This can be accomplished via mechanisms on the Completion property exposed by ActionBlock<T>. The Completion property is a property hanging off of one of the core interfaces implemented by all TPL Dataflow blocks called IDataFlowBlock. As such, all TPL Dataflow blocks, including ActionBlock<T> expose the IDataFlowBlock.Completion property. Using the IDataflowBlock.Completion.Wait() statement we can wait for processing to complete after we've submitted all of our data.

Let's illustrate this in an example. In this example we'll use IDataflowBlock.Completion.Wait to wait for processing to complete. One interesting thing to note is that the Wait method will wait forever because it will never know that ActionBlock<T> is no longer receiving data. To circumvent this, we can tell ActionBlock<T> when we've finished posting data by issuing ActionBlock<T>'s base interface IDataflowBlock.Complete method. This will notify ActionBlock<T> that it will never be giving it more data, so once it's finished processing its queue it can shut itself down, thus notifying us that it has completed processing by releasing the Completion.Wait() call that our main thread is blocking on.

class Program

{

    static void Main(string[] args)

    {

        var action = new ActionBlock<int>(i =>

        {

            Thread.Sleep(500);

            Console.WriteLine(i);

        }, new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 4 });

        for (int i = 0; i < 10; i++) action.Post(i);

        action.Complete();

        Console.WriteLine("Posting Complete!");

        action.Completion.Wait();

    }

}

Besides IDataflowBlock.Complete and IDataflowBlock.Completion, IDataflowBlock also provides a Fault completion mechanism as well. Say for instance we are preprocessing and validating the incoming data before we queue it to our ActionBlock<T> in the example above. If we encounter invalid data during processing we may want to issue a IDataflowBlock.Fault() telling the ActionBlock<T> that we've encountered an invalid state and that is should discontinue processing.

In our next post we will pick up where we left off in our previous post and describe additional dataflow blocks available in TPL Dataflow.

Note: If you want to try these examples out make sure to add a reference to the new System.Threading.Tasks.DataFlow library in your project. Keep in mind; by default the Visual Studio Async CTP bits will be installed in your My Documents library under Microsoft Visual Studio Async CTP, so make sure to browse to that location under the Samples folder to add a reference to the System.Threading.Tasks.DataFlow assembly.

In our previous post, Async CTP - TPL DataFlow, we introduced TPL Dataflow a new library that was released for preview as part of the Microsoft Visual Studio Async CTP. In that post we introduced and described the capabilities that that library provides for building data parallelism into your .NET applications. In this post and the next several posts, we'll expand further on the TPL Dataflow capabilities and provide some examples for using the TPL Dataflow blocks in your applications.

There are several different types of dataflow blocks in the TPL Dataflow library that let you solve a variety of processing scenarios (Join, Transform, etc). One such type is ActionBlock<T>. If you recall from our previous post (Async CTP - TPL DataFlow) where we introduced TPL Dataflow, ActionBlock<T> is an ISourceBlock that accepts and buffers incoming data and performs some action on that data.

You provide ActionBlock<T> with a delegate that will act on the data being posted to it. That delegate executes on every piece of data, in order, that is provided to ActionBlock<T>. The data is posted to ActionBlock<T> via a Post method. The Post method gives ActionBlock<T> data which it will buffer up and internally start spinning tasks to process that data as fast as it can. We can bet illustrate this with an example.

The example below illustrates configuring an ActionBlock<T> with a delegate that simply writes integers out to the console in the order received. Outside ActionBlock<T> is a simple loop that posts integers to it for processing. To make it a little bit easier to see what's going if you try this yourself, we'll introduce a Thread.Sleep so that we can see when and in what order things complete.

class Program

{

    static void Main(string[] args)

    {

        var action = new ActionBlock<int>(i =>

            {

                Thread.Sleep(500);

                Console.WriteLine(i);

            });

        for (int i = 0; i < 10; i++) action.Post(i);

        Console.WriteLine("Posting Complete!");

        Console.ReadLine();

    }

}

There are a couple of interesting things to note from the output. First, you can see that almost immediately the 10 integers were posted to ActionBlock<T>, denoted by the "Posting Complete!" message displayed to the console. The second and more important thing is that the data is being processed in order. This is a key feature of TPL DataFlow. By default, ordering is maintained. This also means that while data is being processed asynchronously, it is also processed sequentially, guaranteeing the order of processing. In other words the data is being processed with a maximum degree of parallelism of 1. What this is telling us is that, in this scenario, the default behavior is not actually to process the input in parallel, but to process it in order, thus guaranteeing the order of processing. Since processing in parallel means processing concurrently and most likely, if not definitely means that things could get processed out of order.

That said there are certain processing scenarios where we are willing to give up on some of these ordering guarantees. ActionBlock<T> is configurable via a mechanism known as ExecutionDataflowBlockOptions that control the degree of parallelism that occurs. The ExecutionDataflowBlockOptions type is essentially an options bag for controlling how ActionBlock<T> will act. One of the options available on ExecutionDataflowBlockOptions is MaxDegreeOfParallelism which, among other things, allows you to specify the maximum number of parallel tasks to execute concurrently on the incoming data. By default this option is set to 1, thus guaranteeing the order of processing.

In this next example we illustrate this by increasing the MaxDegreeOfParallelism to 4. Here the task scheduler will create up to 4 parallel tasks to concurrently process the incoming data.

class Program

{

    static void Main(string[] args)

    {

        var action = new ActionBlock<int>(i =>

            {

                Thread.Sleep(500);

                Console.WriteLine(i);

            }, new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 4 });

        for (int i = 0; i < 10; i++) action.Post(i);

        Console.WriteLine("Posting Complete!");

        Console.ReadLine();

    }

From the output we can see that the data was processed out of order due to the fact that ActionBlock<T> scheduled multiple tasks concurrently to process the incoming data. It is important to note that the MaxDegreeOfParallelism option is not specifying the actual number of tasks to use to process the data, but it is setting a limit on the number of tasks to use. The actual number of tasks used is controlled by the task scheduler, which incidentally can be changed via ExecutionDataflowBlockOptions. There are a variety of ExecutionDataFlowBlockOptions available for controlling how the TPL Dataflow blocks will work. We'll try to get to those in future posts.

ActionBlock<T> is just one example of data flow that exists in TPL Dataflow. However, there are others that further compliment the ability to easily perform parallel processing in your applications. Look for future postings that further describe these blocks as well as the other parallel computing capabilities of the TPF Dataflow library. In our next post we'll diverge slightly and talk about task completion and how we can control that use mechanisms provided to you by IDataflowBlock.

Many of you are probably already familiar with the parallel processing capabilities offered by the Task Parallel Library (TPL) that was first introduced in the .NET Framework 4. In .NET 4 TPL provided some core building blocks and algorithms for doing parallel computing in your .NET applications. Now that the development for the next version of the .NET platform is well under way, it is no surprise that Microsoft is looking for new ways to improve upon a good thing. One such update which should make the final cut is yet another way for developers to build highly responsive code using the new Async capabilities of .NET (see Andrew Troelson's Async CTP Anyone?), currently available as a community technology preview known as the Async CTP. One component of the Async CTP is TPL Dataflow, a library for doing agent/actor based data processing using parallel techniques.

The TPL Dataflow library was first released for preview as part of the Microsoft Visual Studio Async CTP, and it is also available in a separate TPL Dataflow CTP, both of which you can download from http://msdn.microsoft.com/vstudio/async. Both the Async CTP and TPL Dataflow CTP bits can be installed and used with Visual Studio 2010 SP1.

Typically when you did parallel programming with .NET 4 you were being proactive. It was usually the case that you had some data and you wanted to perform some computation on that data. For example, you may have a range of data that you wanted to perform some computation on, or you might need to filter your data in such a way that it was useful to you in your application. To solve these problems in .NET 4 with the Task Parallel Library you might have used Parallel.For or Parallel.Foreach, or possibly even a PLINQ query to iterate through the data and take advantage of the parallel constructs provided to you by TPL. In all these cases is was typically data first, then computations on that data later using primitives for tasks and data parallelism provided to you by the Task Parallel Library. What was missing was the ability to be reactive. Essentially the ability to setup your computation framework first, and then react to the data as it is coming in.

This reactive method of processing data is commonly referred to as dataflow parallelism. Essentially what you're doing is creating computational networks through which data can flow. Agent-based and actor-based message passing patterns like the producer/consumer pattern follow this reactive model. With the introduction of TPL Dataflow you will be able to use the parallel programming paradigms in your own .NET applications to build reactive dataflow networks. This is not to say that you couldn't have solved these problems in .NET 4 using Task Parallel Library, however it would have involved quite a bit of code to manage buffering data, scheduling tasks and dealing with the inter-process communications required to perform the work. The TPL Dataflow library gives you a lot of the primitives that you need to solve these problems out of the box without having to worry about the minutia required to implement such architectures.

So what is TPL Dataflow? Essentially it is a parallel processing library that exposes primitives for doing in-process messaging passing and processing. TPL Dataflow provides a set of agents, commonly referred to as dataflow blocks, or simply blocks, that contain the infrastructure required to buffer and process your data in an asynchronous and parallel manner. TPL Dataflow provides the infrastructure for being able to build data parallelism into your applications.

At the core of the TPL Dataflow library is its interface hierarchy, pictured below. These interfaces describe the behavior of a dataflow block. At the very top of the hierarchy is IDataflowBlock, defining contract for dealing with the lifetime of a dataflow block. Below IDataflowBlock are three sub-interfaces that define contracts for blocks that can be a source of data, blocks that can be a target for data, and blocks that can be both a source and target for data. ISourceBlock represents a source of data, defining a contract for buffering and receiving data. ITargetBlock represents a target for incoming data, defining a contract for buffering and passing data. Finally IPropagatorBlock represents a block that can be both a source and a target for data, defining a contract for receiving data from sources, possibly transforming that data and propagating the result on to other targets.

With these interfaces you can develop your own blocks to build your own custom dataflow networks, but there are also several built in blocks for the most common scenarios like buffering and propagation of data, acting on data, and even blocks for joining data from multiple sources then buffering and presenting the result to a target.

For buffering and propagation of data TPL Dataflow provides the following blocks:

BufferBlock<T> - buffers incoming data and delivers that data to a target in an asynchronous and parallel fashion.

WriteOnceBlock<T> - accepts one piece of data and delivers that piece of data in a broadcast fashion to all target blocks interested in that piece of data.

BroadcastBlock<T> - accepts multiple pieces of data, buffering them as they arrive and broadcasts each piece of data to all interested target blocks.

For acting on incoming data TPL Dataflow provides the following dataflow blocks:

ActionBlock<Tin> - an ISourceBlock that accepts and buffers incoming data and performs some action on that data.

TransformBlock<Tin, Tout> - an IPropagatorBlock that accepts and buffers incoming data, executes some function on that data and then buffers the output to be consumed by a target.

TransformManyBlock<Tin, Tout> - very similar to TransformBlock, however the relationship to its targets is one-to-many. It buffers and transforms incoming data, then buffers and makes that data available to multiple targets.

For joining and grouping data TPL Dataflow provides the following dataflow blocks:

BatchBlock<T> - An IPropagatorBlock that accepts and buffers incoming data and groups that data into batches (arrays).

JoinBlock<T1, T2> - Accepts and buffers data from two different source blocks, combines data from each source into tuples, then buffers and makes those tuples available to a target.

BatchedJoinBlock<T1, T2> - a combination of BatchBlock and JoinBlock that groups pairs of collections of incoming data into tuples of those collections, buffering and presenting that data as output to a target.

As you can see TPL Dataflow provides you with some significant capabilities for developing parallel dataflow capabilities into your own applications without having to reinvent the infrastructure to do so. This was a quick look at the TPL Dataflow and the interface hierarchy upon which it is build, as well as some of the built in capabilities for building applications that incorporate data parallelism. Watch for future posts where we'll look at some of these built in mechanisms along with examples demonstrating how you can get started using TPL Dataflow.

If you are interested in looking at the current state of this new technology, here are some helpful links: 

• Whitepaper: Asynchrony in .NET
• Async CTP (preview bits and sample code)