C# Corner

Exploring .NET 4.5 Parallel Dataflow - Part 2

In Part 2 of this three-part series on dataflow programming with the Task Parallel Library, Eric Vogel shows you how to create a Windows 8 application that uses a composite parallel data flow.

• By Eric Vogel
• 09/19/2012

In Part 1 of this series on building Windows 8 applications with the Task Parallel Library (TPL) dataflow components in the .NET Framework, I covered how to use the ActionBlock and the TransformBlock dataflow blocks. This time, I'll take it a step further and show you how to link dataflow blocks, to create more complex parallel data flows. A common usage pattern is the producer/consumer scenario. By linking dataflow blocks together, one block can post a message that is then pumped through one or many dataflow blocks that further process the data.

To demonstrate this concept, let's look at a hypothetical data flow for an application. Say you're reading temperatures from a sensor in regular sub-second intervals. The data needs to be displayed as it's read in real-time. In addition, you need to be able to apply some post processing to the raw temperatures such as formatting for Fahrenheit, and comparisons to the average temperature for the current date.

This scenario tackles a few common dataflow issues. For one, the data needs to be broadcast to multiple sources. Secondly, the data must be read by multiple sources in order to be further processed and formatted. Luckily, the Task Parallel Dataflow (TDF) library includes the BroadcastBlock, which makes implementing this scenario straightforward. Without further ado, let's get down to the details.

Open up Visual Studio 2012 Release Candidate and create a C# Metro style App. First, open up MainPage.xaml and use the XAML from the root Grid element in Listing 1. The UI is fairly simple. Three sets of stack panels are associated with the three buttons on the page.

Broadcasting Data
For this sample application, you'll need to have the Parallel Dataflow NuGet package installed. Refer to Part 1 for installation instructions. Now that the TDF package is installed, open up MainPage.xaml.cs and add the following using statements:

 using System.Threading.Tasks;
 using System.Threading.Tasks.Dataflow;

Then add the BroadcastBlock to the MainPage class:

 BroadcastBlock<int> _broadcaster;

The BroadcastBlock will be responsible for transferring temperature data to the various TransformBlock objects for further processing. Once a Transform block has processed and formatted a piece of datum, it'll pump the datum to an ActionBlock for UI display.

Next, instantiate the _broadcaster block in the OnNavigatedTo event of the page:

 protected override void OnNavigatedTo(NavigationEventArgs e)
{
    _broadcaster = new BroadcastBlock<int>(x => x);
}

Now setup the click event handler for the BroadCast button. Within the event, an ActionBlock is created that will display a message via the Message TextBlock on the page:

private async void BroadCast_Click(object sender, RoutedEventArgs e)
{
    ActionBlock<int> simpleDisplayer =
CreateUiUpdateActionBlock<int>(Message);
    _broadcaster.LinkTo(simpleDisplayer);
    await BroadCastData();
}

The CreateUiUpdateActionBlock<T> method is a simple helper function that creates an ActionBlock to set the Text property of a TextBlock element on the UI thread:

private ActionBlock<T> CreateUiUpdateActionBlock<T>(TextBlock element)
{
    return new ActionBlock<T>(x =>
        element.Text = x.ToString(),
          new ExecutionDataflowBlockOptions() { TaskScheduler =
TaskScheduler.FromCurrentSynchronizationContext() });
}

Next, the simpleDisplayer ActionBlock is linked to the _broadcaster block through the LinkTo method:

 _broadcaster.LinkTo(simpleDisplayer);

Now, there's a lot of power in that link statement. Whenever any data is posted to the _broadcaster block, it will immediately be propagated to the simpleDisplayer block asynchronously.

Next, I'll generate and post some randomized temperature data to the _broadcaster block by awaiting the BroadCastData method:

await BroadCastData();
private async Task BroadCastData()
{
    Random r = new Random((int)System.DateTime.Now.Ticks);
    int temp = 0;

    for (int i = 0; i < 1000; i++)
    {
        await Task.Delay(125);
        temp = r.Next(60, 80);
        await _broadcaster.SendAsync(temp);
    }
}

I've added a 125-msec delay between temperature datum postings to more accurately simulate reading from a sensor.

Post Processing Data
Now that the application is correctly broadcasting and displaying temperature data, let's get started on the second requirement. When the Transform button is clicked, the temperature should be displayed in the Fahrenheit format. To accomplish this task, the TransformBlock is ideal because it can receive a raw integer temperature and format it to a Fahrenheit formatted string:

 TransformBlock<int, string> _formatter = new TransformBlock<int, string>(x =>
 String.Format("{0:G2}°F", x));

Once the data has been cleaned up, it can be displayed to the user via an ActionBlock:

ActionBlock<string> transformedDisplayed =         
CreateUiUpdateActionBlock<string>(TransformedMessage);

The last step is to link all of the dataflow blocks together so that the _broadcaster sends data to the formatter, which displays data via the transformDisplayed block:

 _broadcaster.LinkTo(formatter);
 formatter.LinkTo(transformedDisplayed);

private void Transform_Click(object sender, RoutedEventArgs e)
{
    TransformBlock<int, string> formatter = new TransformBlock<int, string>(x =>
    String.Format("{0:G2}°F", x));
    ActionBlock<string> transformedDisplayed =
CreateUiUpdateActionBlock<string>(TransformedMessage);
    _broadcaster.LinkTo(formatter);
    formatter.LinkTo(transformedDisplayed);
}

Now, let's implement the last requirement for displaying a formatted Fahrenheit delta temperature between a read temperature and the average temperature for the day.

The processing will occur in the PostProcess button click event handler as shown in Listing 2 .

Computing the delta average temperature is easily accomplished via a TransformBlock:

 const int average = 75;
 TransformBlock<int, int> averageDelta = new TransformBlock<int, int>(x => x - average);

Now another TransformBlock is created to format the delta temperature to be displayed in degrees Fahrenheit. In addition, a '+' or '-' is pretended to the temperature to indicate the change:

TransformBlock<int, string> formatPost = new TransformBlock<int, string>(x => {
  string preFix = string.Empty;
  if (x > 0) preFix = "+";
  return String.Format("{0}{1:G2}°F", preFix, x);
 });

Finally, an ActionBlock is created to update the text of the PostProcessMessage TextBlock on the page, and the data flow blocks are linked together:

ActionBlock<string> postProcessDisplayer =
CreateUiUpdateActionBlock<string>(PostProcessMessage);
_broadcaster.LinkTo(averageDelta);
averageDelta.LinkTo(formatPost);
formatPost.LinkTo(postProcessDisplayer);

You should now be able to run the completed application shown in Figure 1.

Figure 1. Completed sample application

As you can see, the Task Parallel Dataflow library is very versatile for implementing complex parallel data flows. Through dataflow block composition, a myriad of problems can be solved in a simple, efficient and elegant manner. Stay tuned for Part 3 in this series on dataflow programming with the Task Parallel Library dataflow components in .NET Framework 4.5 to learn how to create custom dataflow blocks.

• Get Code Download