Transform Block

Let us first consider the piece of code in Listing 10-4. The purpose of this code is to turn a color-based image into a grayscale image. The code is expected to initially run on the UI thread; so as to not block the UI thread, it creates a task to run the image processing and, once completed, fires an event on the UI thread to indicate it is done.

Listing 10-4.  Asynchronously Converts Images to WPF Grayscale Bitmap Images

public class ImageProcessor
{
  public event EventHandler<ProcessedImageEventArgs> ProducedGrayScaleImage = delegate { };
 
  public void ProcessFile(string filename)
  {
    Task.Run(() =>LoadAndToGrayScale(filename))
        .ContinueWith(toGrayTask =>ProducedGrayScaleImage(this,
                                    new ProcessedImageEventArgs(toGrayTask.Result)),
                   TaskScheduler.FromCurrentSynchronizationContext());
 
         
  }
  private static  BitmapSource LoadAndToGrayScale(string path)
  {
      var img = new BitmapImage(new Uri(path));
      return ToGrayScale(img);
  }
         
  private static  BitmapSource ToGrayScale(BitmapSource bitmapSource)
  {
        . . .
  }
}

Instead of explicitly using tasks, you could refactor to use two dataflow blocks, as per Figure 10-7.

Items can be posted to the transform block in the same way as you previously posted work to an action block. Unlike the action block, the transform block has a responsibility not just to consume but to produce. For the transform block to work, you need to supply the transformation logic. This is achieved by supplying a Func<TInput,TOuput> to the block. When data are posted to the block, the supplied function is invoked to produce the output. As with the action block, the transform block will have by default only a single task executing at any one time to perform the transformation logic.

By default ActionBlock<T> tasks will use the default task scheduler, which is not what we want, since this block’s responsibility is to fire the event on the UI thread. To solve this, create the ActionBlock<T> specifying that you want it to use the task scheduler associated with the current synchronization context.

Once both the blocks have been created, you need to link them up. To do this, there is a LinkTo method on the TransformBlock<TInput,TOutput>. This LinkTo method takes as an argument any block that has an input type of TOutput. The LinkTo method is actually defined on the ISourceBlock<TOutput> interface, which is implemented by any block that can be a source (i.e., has an output). In a similar fashion, any block that has input, implements ITargetBlock<T>. The LinkTo method returns an IDisposable object, which when disposed removes the link between the blocks.

Last, modify the ProcessFile method simply to post the requested file to the transform block, and return in the hope that someday an event will fire with the converted image (see Listing 10-5).

Listing 10-5.  Dataflow for Converting Image to Grayscale

public class ImageProcessor
{
  public event EventHandler<ProcessedImageEventArgs> ProducedGrayScaleImage = delegate { };
 
  private TransformBlock<string, BitmapSource> loadAndToGrayBlock;
  private ActionBlock<BitmapSource> publishImageBlock;
        
  public ImageProcessor()
  {
    loadAndToGrayBlock = new TransformBlock<string, BitmapSource>(
                           (Func<string, BitmapSource>)LoadAndToGrayScale);
    publishImageBlock = new ActionBlock<BitmapSource>((Action<BitmapSource>) PublishImage,
                new ExecutionDataflowBlockOptions()
                {
                    TaskScheduler = TaskScheduler.FromCurrentSynchronizationContext()
                });
    loadAndToGrayBlock.LinkTo(publishImageBlock);
  }
 
  public void ProcessFile(string filename)
  {
   loadAndToGrayBlock.Post(filename);
  }
 
  private void PublishImage(BitmapSource img)
  {
     ProducedGrayScaleImage(this,new ProcessedImageEventArgs(img));
  }
 
  private static  BitmapSource LoadAndToGrayScale(string path)
  {
    var img = new BitmapImage(new Uri(path));
    return ToGrayScale(img);
  }
  private static  BitmapSource ToGrayScale(BitmapSource bitmapSource)
  {
     . . .
  }
}

Listing 10-5 contains the refactored code that now utilizes blocks to provide concurrency as opposed to using tasks explicitly. Instead of thinking about threads of execution, think about blocks of concurrent execution, stimulated by asynchronous messages.

Transform Many Block

The ImageProcessor class currently allows you to process individual files, but what if you wanted to process an entire directory? Simple enough—just iterate through all the image files from a directory search and post the file, as per Listing 10-6. While this will obviously work, you are in fact missing an opportunity to keep things fast and fluid: rather than performing the directory search on the UI thread, you could encapsulate the directory search functionality inside a new block. The UI thread will then just simply post the directory start point to that block, and return to UI processing.

Listing 10-6.  Process Multiple Files

public void ProcessDirectory(string dir)
{
   DirectoryInfo directory = new DirectoryInfo(dir);
   foreach (FileInfo file in directory.GetFiles("*.jpg"))
   {
   loadAndToGrayBlock.Post(file.FullName);
   }
}

The TransformBlock<TInput,TOuput> block produces as many outputs as it receives inputs. You now need a block that takes a directory path as an input and potentially can produce many image files, thus producing an imbalance of inputs to outputs. You could make the directory block produce, say, a List<string>, but that would batch as opposed to stream. A more scalable approach would be to use a TransformManyBlock<TInput,TOutput>; this block does not simply take a Func<TInput,TOutput> but a Func<TInput,IEnumerable<TOutput>>. The code for the block produces an IEnumerable<TOutput>, potentially offering many results that can then be passed onto a linked block. The Dataflow framework consumes the IEnumerable<TOutput>, making the output as it becomes available to any linked blocks. Figure 10-8 shows the new block topology, allowing directories and files to be processed concurrently.

To implement this new topology, you need to create an instance of a TransformManyBlock and provide a method on the ImageProcessor class to take a directory and post it to the block (Listing 10-7).

Listing 10-7.   TransformMany

public class ImageProcessor
{
  public event EventHandler<ProcessedImageEventArgs> ProducedGrayScaleImage = delegate { };
 
  private TransformManyBlock<string, string> imageCollectionBlock;
  private TransformBlock<string, BitmapSource> loadAndToGrayBlock;
  private ActionBlock<BitmapSource> publishImageBlock;
        
  public ImageProcessor()
  {
    imageCollectionBlock = new TransformManyBlock<string, string>(
                    (Func<string, IEnumerable<string>>) FindImagesInDirectory);
 
    loadAndToGrayBlock = new TransformBlock<string, BitmapSource>(
                                 (Func<string, BitmapSource>)LoadAndToGrayScale);
 
    publishImageBlock = new ActionBlock<BitmapSource>((Action<BitmapSource>) PublishImage,
                new ExecutionDataflowBlockOptions()
                {
                    TaskScheduler = TaskScheduler.FromCurrentSynchronizationContext()
                });
 
     imageCollectionBlock.LinkTo(loadAndToGrayBlock);
     loadAndToGrayBlock.LinkTo(publishImageBlock);
   }
    
   public void ProcessFile(string filename)
   {
     loadAndToGrayBlock.Post(filename);
   }
 
   public void ProcessDirectory(string dir)
   {
    imageCollectionBlock.Post(dir);
   }
 
   private IEnumerable<string> FindImagesInDirectory(string dir)
   {
     var directory = new System.IO.DirectoryInfo(dir);
     return directory
                .GetFiles("*.jpg")
                .Select(file => file.FullName);
   }
   private void PublishImage(BitmapSource img)
   {
      ProducedGrayScaleImage(this, new ProcessedImageEventArgs(img));
   }
   private static  BitmapSource LoadAndToGrayScale(string path)
   {
      var img = new BitmapImage(new Uri(path));
      return ToGrayScale(img);
   }
   private static  BitmapSource ToGrayScale(BitmapSource bitmapSource)
   {
     . . .
   }
}

Now that you have a working network of blocks, you can start to consider further concurrency. At present you are only using a single task to turn images into grayscale; of all the blocks this is the one that will take the time. The grayscale processing of each image is isolated—you don’t need to care about the order in which the images appear, and therefore it seems logical that you should assign multiple tasks to this block. Listing 10-8 shows the necessary change to the ImageProcessor class to enable concurrent processing of images.

Listing 10-8.  Many Tasks Performing Image Conversion

loadAndToGrayBlock = new TransformBlock<string, BitmapSource>(
                                    (Func<string, BitmapSource>)LoadAndToGrayScale,
                       new ExecutionDataflowBlockOptions()
                       {
                         MaxDegreeOfParallelism = 4,
                       });

While you easily could have written the image processor using tasks, one advantage to using TPL Dataflow is being able to control the number of threads for a given role. In a large application you may very well want to take finer control over how compute resources are being shared across the entire application, and not just have them be shared evenly, as would be the case if everything were throttled exclusively by the thread pool.

Linking to Multiple Targets

So far we have only considered linking a source block to a single target block. Consider the situation where, based on a given input, data are fetched from a database, and for performance reasons the database is mirrored, allowing data to be fetched from either database. You should therefore consider building a dataflow as described in Figure 10-9. Assuming that if Server A is busy, Server B will be offered the query, then if both blocks are busy the previous block will wait until either becomes available. Just simply linking up multiple targets in this way, unfortunately, will not give you this effect. Most blocks by default have associated with them an unbounded queue that allows them to receive messages even if they are busy. This is known as greedy behavior, and the level of greediness can be controlled by setting the capacity of the queue. To make a block non-greedy, simply set the queue length to 1. Greedy blocks can therefore accept messages even if they are busy processing a previous message. Non-greedy blocks will refuse messages while they are busy processing message(s). If a message is refused by one block, the next linked block will be offered the message. If all blocks refuse the message, the first block to become available to process the message will do so (see Listing 10-9).

Listing 10-9.  Greedy and Non-greedy Blocks

class Program
{
   static void Main(string[] args)
   {
     var greedy = new ExecutionDataflowBlockOptions();
 
     var nonGreedy = new ExecutionDataflowBlockOptions()
     {
        BoundedCapacity = 1
     };
 
     ExecutionDataflowBlockOptions options = greedy;
 
     var firstBlock = new ActionBlock<int>(i => Do(i,1,2), options);
     var secondBlock = new ActionBlock<int>(i => Do(i,2,1), options);
     var thirdBlock = new ActionBlock<int>(i => Do(i,3,2), options);
 
     var transform = new TransformBlock<int,int>(i=>i*2);
 
     transform.LinkTo(firstBlock);
     transform.LinkTo(secondBlock);
     transform.LinkTo(thirdBlock);
 
     for (int i = 0; i <= 10; i++)
     {
       transform.Post(i);
     }
     Console.ReadLine();
  }
  private static void Do(int workItem , int nWorker, int busyTimeInSeconds)
  {
     Console.WriteLine("Worker {0} Busy Processing {1}",nWorker,workItem);
     Thread.Sleep(busyTimeInSeconds * 1000 );
     Console.WriteLine("Worker {0} Done",nWorker);
  }
}

Running the code in Listing 10-9 will show greedy behavior, as shown in Figure 10-10. Here Worker 1 has an unbounded queue, so while it is busy it can still accept requests, which it eventually processes.

Modifying the code in Listing 10-9 to use the nonGreedy configuration will produce the output shown in Figure 10-11. With a capacity of 1, each worker refuses any more work until it is idle, resulting in messages being acted upon by the first available worker.

public interface ILedgerEntry
{
   bool IsCredit { get; }
   bool IsDebit { get; }
};
 
public class CsvImporter
{
  private TransformManyBlock<string, object[]> databaseQueryBlock;
  private TransformManyBlock<object[], ILedgerEntry> rowTGoLedgerBlock;
  private ActionBlock<ILedgerEntry> debitBlock;
  private ActionBlock<ILedgerEntry> creditBlock;
  private ActionBlock<ILedgerEntry> unknownLedgerEntryBlock;
 
  public CsvImporter()
  {
    databaseQueryBlock = new TransformManyBlock<string, object[]>(
                     (Func<string, IEnumerable<object[]>>) ExecuteQuery);
    rowTGoLedgerBlock = new TransformManyBlock<object[], ILedgerEntry>(
                     (Func<object[], IEnumerable<ILedgerEntry>>) MapDatabaseRowToObject);
    debitBlock = new ActionBlock<ILedgerEntry>((Action<ILedgerEntry>) WriteDebitEntry);
    creditBlock = new ActionBlock<ILedgerEntry>((Action<ILedgerEntry>) WriteCreditEntry);
 
    unknownLedgerEntryBlock = new ActionBlock<ILedgerEntry>(
                                     (Action<ILedgerEntry>)LogUnknownLedgerEntryType);
 
    databaseQueryBlock.LinkTo(rowTGoLedgerBlock);
    rowTGoLedgerBlock.LinkTo(debitBlock, le => le.IsDebit); //if IsDebit
    rowTGoLedgerBlock.LinkTo(creditBlock, le => le.IsCredit); // else if IsCredit
    rowTGoLedgerBlock.LinkTo(unknownLedgerEntryBlock); // else
  }
 
 
  public void Export(string connectionString)
  {
    databaseQueryBlock.Post(connectionString);
  }
        
  private IEnumerable<object[]> ExecuteQuery(string arg){ yield break; }
  private IEnumerable<ILedgerEntry> MapDatabaseRowToObject(object[] arg) { yield break;}
  private void WriteDebitEntry(ILedgerEntry debitEntry) { }
  private void WriteCreditEntry(ILedgerEntry creditEntry) { }
  private void LogUnknownLedgerEntryType(ILedgerEntry obj){}
}

private static IEnumerable<string> GetFiles(string path)
{
  DirectoryInfo dir = new DirectoryInfo(path);
  foreach (FileInfo file in dir.GetFiles())
  {
     yield return dir.FullName;
  }
  foreach (DirectoryInfo subDir in dir.GetDirectories())
  {
    foreach (string file in GetFiles(subDir.FullName))
    {
      yield return file;
    }
  }
}

public class DirectoryWalker
{
   private ActionBlock<string> fileActionBlock;
   private TransformManyBlock<string, string> directoryBrowseBlock;
         
   public DirectoryWalker(Action<string> fileAction)
   {
     directoryBrowseBlock = new TransformManyBlock<string, string>(
                            (Func<string, IEnumerable<string>>)(GetFilesInDirectory));
     fileActionBlock = new ActionBlock<string>(fileAction);
 
   directoryBrowseBlock.LinkTo(directoryBrowseBlock, Directory.Exists);
   directoryBrowseBlock.LinkTo(fileActionBlock);
 }
 
 public void Walk(string path)
 {
   directoryBrowseBlock.Post(path);
 }
 
 private IEnumerable<string> GetFilesInDirectory(string path)
 {
   var dir = new DirectoryInfo(path);
   return dir.EnumerateFileSystemInfos().Select(fi => fi.FullName);
 }
}
. . .
var walker = new DirectoryWalker(Console.WriteLine);
walker.Walk(@"C:	emp");
Console.ReadLine();

Propagating Completion

As already discussed, dataflow blocks are often not used as islands but as part of a larger network of blocks. Listing 10-15 shows some example code that contains two linked blocks. Ten items are posted to the first block; then the block is told there will be no more, and the main thread then waits for the first block to complete. Once the block is completed, the application terminates. Since the second block takes 0.5 seconds to process each block, the main thread will terminate before the second block completes processing its first item.

Listing 10-15.  Premature Completion

class Program
{
  static void Main(string[] args)
  {
    var firstBlock = new TransformBlock<int, int>(i => i*2);
    var secondBlock = new ActionBlock<int>( i =>
    {
      Thread.Sleep(500);
      Console.WriteLine(i);
    });
 
   firstBlock.LinkTo(secondBlock);
 
   for (int i = 0; i < 10; i++)
   {
     firstBlock.Post(i);
   }
   firstBlock.Complete();
   firstBlock.Completion.Wait();
  }
}

To resolve this premature termination issue, you could add two additional lines after the completion logic for the first block:

secondBlock.Complete();
secondBlock.Completion.Wait();

It is important to ensure that you don’t complete the second block until the first block has completed, since the second block will not accept any further input once it has been told to complete. As you can probably imagine this would get tedious as you increased the number of blocks. An alternative is to allow completion to automatically flow to each of its linked blocks as a block completes. Listing 10-16 shows refactored code that now utilizes completion propagation. Automatic propagation is requested as part of linking one block to another. Once the source block has been marked as complete and it has processed all queued items, it informs the linked block that it is now complete. This now greatly simplifies the logic necessary to determine when all inputs have been fully processed by the dataflow network.

Listing 10-16.  Automatic Completion Propagation

class Program
{
  static void Main(string[] args)
  {
    var firstBlock = new TransformBlock<int, int>(i => i*2);
    var secondBlock = new ActionBlock<int>( i =>
    {
       Thread.Sleep(500);
       Console.WriteLine(i);
    });
 
    firstBlock.LinkTo(secondBlock , new DataflowLinkOptions(){PropagateCompletion = true});
     
    for (int i = 0; i < 10; i++)
    {
      firstBlock.Post(i);
    }
 
     firstBlock.Complete();
     secondBlock.Completion.Wait();
  }
}

Error Handling

So far we have only considered the happy side of programming, when everything just works. In the real world we have to consider error handling, and in .NET this takes the form of exceptions. Dataflow blocks are fundamentally blocks of code, and as such they can throw exceptions. Consider the code in Listing 10-17. The first two work items should work fine. The third item will cause a divide by zero exception and hence will not produce any output. But what about the fourth item? There are two possibilities: the block swallows the exception and processes the fourth item, or it refuses to process any more items. The actual behavior is the latter; once a block experiences an unhandled exception, it places the block into a faulted state and refuses to process or receive any more messages.

Listing 10-17.  Unhandled Divide by Zero Exception

private static void Main(string[] args)
{
  var divideBlock = new ActionBlock<Tuple<int, int>>(
            (Action<Tuple<int, int>>) (input => Console.WriteLine(input.Item1/input.Item2)));
 
  divideBlock.Post(Tuple.Create(10, 5));
  divideBlock.Post(Tuple.Create(20, 4));
  divideBlock.Post(Tuple.Create(10, 0));
  divideBlock.Post(Tuple.Create(10, 2));
 
   Console.ReadLine();
}

Catching the exception and dealing with it inside the block is obviously one remedy, and makes perfect sense if the exception doesn’t indicate that the overall dataflow network is now effectively broken, as shown in Listing 10-18.

Listing 10-18.  Handling Nonfatal Exceptions Inside the Block

private static void Main(string[] args)
{
  var divideBlock = new ActionBlock<Tuple<int, int>>((Action<Tuple<int, int>>)
        delegate(Tuple<int, int> pair)
        {
           try
           {
              Console.WriteLine(pair.Item1/pair.Item2);
           }
           catch (DivideByZeroException)
           {
             Console.WriteLine("Dude, can't divide by zero");
           }
         });
  divideBlock.Post(Tuple.Create(10, 5));
  divideBlock.Post(Tuple.Create(20, 4));
  divideBlock.Post(Tuple.Create(10, 0));
  divideBlock.Post(Tuple.Create(10, 2));
 
   Console.ReadLine();
}

In cases where the underlying exception implies that it is not safe or desirable to proceed with any more messages, then the recovery logic is probably best suited outside the dataflow network. The Completion task discussed earlier provides a means for externally observing if a block has received an unhandled exception. If the status of the Completion task is in a state of Faulted, then this indicates the block is in a faulted state and will perform no further processing. Listing 10-19 demonstrates the wiring up of a continuation in the case of the block ending in a faulted state.

Listing 10-19.  Externally Handling a Block Exception

private static void Main(string[] args)
{
  var divideBlock = new ActionBlock<Tuple<int, int>>(
            (Action<Tuple<int, int>>) (input => Console.WriteLine(input.Item1/input.Item2)));
 
  divideBlock.Post(Tuple.Create(10, 5));
  divideBlock.Post(Tuple.Create(20, 4));
  divideBlock.Post(Tuple.Create(10, 0));
  divideBlock.Post(Tuple.Create(10, 2));
 
  divideBlock
   .Completion
   .ContinueWith(dbt =>
   {
      Console.WriteLine("Divide block failed Reason:{0}",
                          dbt.Exception.InnerExceptions.First().Message);
   }, TaskContinuationOptions.OnlyOnFaulted);
 
   Console.ReadLine();
}

So far we have examined error handling at the individual block level. Externally handling errors at the block level can become tedious. A more convenient approach may be to just handle the final outcome of the dataflow network, in the same way we often put a try/catch around a block of sequential processing. When linking blocks together, the setting of the PropagateCompletion flag as part of the DataflowLinkOptions will not only propagate successful completion but also propagate errors. Listing 10-20 implements a two-block network, where the error originates in the first block but it is the Wait on the second block that receives the error.

Listing 10-20.  Propagating Errors

private static void Main(string[] args)
{
  var divideBlock = new TransformBlock<Tuple<int, int>, int>(
                              input => input.Item1 / input.Item2);
  var printingBlock = new ActionBlock<int>((Action<int>)Console.WriteLine);
 
  divideBlock.LinkTo(printingBlock, new DataflowLinkOptions() { PropagateCompletion = true });
 
  divideBlock.Post(Tuple.Create(10, 5));
  divideBlock.Post(Tuple.Create(20, 4));
  divideBlock.Post(Tuple.Create(10, 0));
  divideBlock.Post(Tuple.Create(10, 2));
 
  divideBlock.Complete();
 
 try
 {
    printingBlock.Completion.Wait();
 }
 catch (AggregateException errors)
 {
   foreach (Exception error in errors.Flatten().InnerExceptions)
   {
     Console.WriteLine("Divide block failed Reason:{0}", error.Message);
   }
 }
}

Error handling is a fundamental part of any application, and dataflow networks are no exception to this rule. When building a dataflow network, consider where to place the error handling.

• If the exception can be handled internally, keeping the dataflow network integrity intact, do so.
• If the exception implies the dataflow network integrity is now compromised, consider handling the exception externally, via a continuation.
• Propagating completion often simplifies exception handling, allowing a single piece of error-handling logic.

Ignoring error handling can easily result in a dataflow network that sits idle, not receiving any new messages—a nightmare to debug.

Cancellation

You have seen that dataflow networks can be asked to complete by stating no further inputs will be supplied to the block. The block will still continue running while it has more queued work to process and has not entered a Faulted state. If you then wait on the block, you can be confident that all requests have been successfully acted upon. Waiting for the network to fully process all outstanding items may not be completely desirable. A quicker form of termination may be required; obviously simply ending the process would be one such technique, potentially leaving external resources in an inconsistent state (e.g., an XML file not closed off correctly). A halfway point between fully completing all requests and ending abruptly is to make use of TPL's CancellationTokenSource and CancellationToken.

When a block is created, it can be supplied with a CancellationToken. This token is observed by the block to determine if the block should cease receiving, acting on, or producing values. Listing 10-21 creates a block and associates a CancellationToken with the block. The block is then posted three items; the processing of each item takes 1 second. If at any time the Enter key is pressed while these items are being processed, the main thread calls cts.Cancel(), thus requesting the block to cancel. If the block has not already completed all the enqueued requests, the block will end in a canceled state.

Listing 10-21.  Cancelling a Block

private static void Main(string[] args)
{
  CancellationTokenSource cts = new CancellationTokenSource();
 
  var slowAction = new ActionBlock<int>( (Action<int>) (
         i =>
              {
                    Console.WriteLine("{0}:Started",i);
                    Thread.Sleep(1000);
                    Console.WriteLine("{0}:Done", i);
              }),
              new ExecutionDataflowBlockOptions() { CancellationToken = cts.Token} );
 
  slowAction.Post(1);
  slowAction.Post(2);
  slowAction.Post(3);
 
  slowAction.Complete();
 
  slowAction
            .Completion
            .ContinueWith(sab => Console.WriteLine("Blocked finished in state of {0}",
                                                   sab.Status));
 
  Console.ReadLine();
  cts.Cancel();
 
  Console.ReadLine();
}

Running the code in Listing 10-21 and pressing Enter while it is processing the second item produces the output shown in Figure 10-16. Once called, the block does not attempt to process Item 3, but it does complete the processing of Item 2. Thus far the cancellation token is only being observed by the block and not the code running inside the block.

The framework can’t just simply decide to abort the code, and so the code inside the block also needs to observe the cancellation token when it is safe to abort. Listing 10-22 shows the necessary changes to the block to make it respond to the cancellation, too.

Listing 10-22.  Cancelling Block and Code

var slowAction = new ActionBlock<int>(
                (Action<int>) (i =>
               {
                    Console.WriteLine("{0}:Started",i);
                    cts.Token.WaitHandle.WaitOne(1000);
                    cts.Token.ThrowIfCancellationRequested();
                    Console.WriteLine("{0}:Done", i);
                }),
                new ExecutionDataflowBlockOptions() { CancellationToken = cts.Token} );

When multiple blocks are linked together with the PropagateCompletion property set to true, canceling a block will then mark the linked blocks as complete and in a state of RanToCompletion, potentially starting a completion chain reaction through the remaining linked blocks. To cause a more rapid shutdown, each block must share the cancellation token.

Buffer Block

Execution blocks have the possibility to buffer items internally. Earlier you saw that you can control the size of that buffer to create non-greedy blocks, blocks that would not consume items unless they could process them immediately. You can use this technique to provide a form of load balancing as shown in Figure 10-17, in which the Server A block will only accept requests if it is not busy, and likewise Server B.

The problem with the approach in Figure 10-17 is that if both Server A and Server B are busy, the transform feeding these two blocks becomes blocked and cannot process any more inputs. So ideally you want to buffer any output from the transform block and then have either Server A or Server B process the output when they are free.

Listing 10-23 shows a simplified example using the topology shown in Figure 10-18. Running the code will demonstrate that the transfer block completes all its processing very quickly and that the two load-balanced processor blocks eventually consume everything in the buffer.

Listing 10-23.  Buffered Load Balanced

static void Main(string[] args)
{
   var nonGreedy = new ExecutionDataflowBlockOptions() {BoundedCapacity = 1};
   var flowComplete = new DataflowLinkOptions() {PropagateCompletion = true};
 
   var processorA = new ActionBlock<int>((Action<int>)( i => Processor("A",i)),nonGreedy);
   var processorB = new ActionBlock<int>((Action<int>)( i => Processor("B",i)),nonGreedy);
   var transform = new TransformBlock<int, int>(i => i*2);
   var buffer = new BufferBlock<int>();
 
   buffer.LinkTo(processorA,flowComplete);
   buffer.LinkTo(processorB,flowComplete);
 
   transform.LinkTo(buffer);
 
  // transform.LinkTo(processorA);
  // transform.LinkTo(processorB);
 
   for (int i = 0; i < 5; i++)
   {
     transform.Post(i);
   }
   transform.Complete();
   transform.Completion.Wait();
   Console.WriteLine("All work buffered ");
 
   Console.ReadLine();
}
private static void Processor( string name , int value)
{
   Console.WriteLine("Processor {0}, starting : {1}",name,value);
   Thread.Sleep(1000);
   Console.WriteLine("Processor {0}, done : {1}",name,value);
}

Modifying Listing 10-23 by uncommenting the transform.LinkTo(processorA) or transform.LinkTo(processorB) and commenting out transform.LinkTo(buffer) will show the effect of load balancing without the buffer. The effect you will observe is that the transform block takes longer to produce all the items as it is waiting for a consumer before publishing the next value.

Buffer blocks are therefore useful when you need a shared buffer across multiple consumers, and each consumer has its own internal constrained buffer. The internal constrained buffer should be set to the same value as the MaxDegreeOfParallelism property for the block in order to ensure a non-greedy block.

Batch Block

Thus far you have seen that transform blocks for a given input can produce one to many outputs. The consumer of these outputs needs to receive the output before the transform block can produce a new item. There will be times when for efficiency or convenience the consumer would rather not deal with each individual output, but would prefer a batch of outputs. Classic use cases for this may be to upload results to a web service or update a database; a round trip per result is inefficient, and batching them means paying the latency cost once for all the items in the batch. Another possible use case would be a dataflow network producing values that need to be used to update the UI, but marshaling back onto the UI thread per request may cause the UI to become sluggish. Marshaling less frequently and performing many updates may result in a more fluid UI. Listing 10-24 shows an example of using buffer blocks to provide sampling of a dense dataflow. The main thread is producing a new value every 10 milliseconds, the averager action block receives the generated values in batches of 100, and then outputs a sample result by simply averaging all the values. If the batch block was marked as complete, a partial batch is propagated.

Listing 10-24.  Sampling by Averaging 100 Items

public static void Main()
{
  int batchSize = 100;
  var batcher = new BatchBlock<int>(batchSize);
 
  var averager = new ActionBlock<int[]>(values => Console.WriteLine(values.Average()));
 
  batcher.LinkTo(averager);
  var rnd = new Random();
  while (true)
  {
    Thread.Sleep(10);
    batcher.Post(rnd.Next(1, 100));
  }
}

Listing 10-24 demonstrates batching based on quantity; it may be more desirable to batch on a time interval basis (e.g., for UI update). This can be achieved with the BatchBlock, too, by setting the batch size to int.MaxValue and then calling TriggerBatch method at a given time interval (see Listing 10-25).

Listing 10-25.  Sampling by Interval

public static void Main()
{
   var batcher = new BatchBlock<int>(int.MaxValue);
   var averager = new ActionBlock<int[]>(values => Console.WriteLine(values.Average()));
 
   batcher.LinkTo(averager);
 
   var timer = new Timer(_ => batcher.TriggerBatch(),null,
                         TimeSpan.FromSeconds(1),
                         TimeSpan.FromSeconds(1));
 
   var rnd = new Random();
   while (true)
   {
     batcher.Post(rnd.Next(1, 100));
    }
 }

BatchBlocks are useful when processing of individual results is too costly, or there is a need to perform a Many-1 transformation.

Broadcast Block

Broadcast blocks differ from all the other blocks in that they can deliver the same message to multiple targets. This allows the building of networks where many blocks receive the same message, and thus can act on it in parallel. Consider the network in Figure 10-19; here data arrive via some external input and are posted to a BroadcastBlock<T>. The broadcast block can then propagate the input to all the linked blocks. If the block is not busy or it has sufficient buffer (greedy), then the input will be propagated. If the block is busy or has insufficient buffer, the input is not propagated. The two consumers in Figure 10-19 both get to see all the messages. Processing the data inside the Process Data block is very quick compared to writing to disk, and therefore it may be that the disk can’t keep up with the input stream. Buffering could be used, but this could easily result in an out of memory exception if the input rate is relentless. Configuring the write data block to be non-greedy means that it writes as much data as it can, and any items that occur while it is busy it will never see. The process data block, on the other hand, is configured to be greedy, so any slight bumps in processing will still result in it receiving all the input. BroadcastBlock<T> therefore doesn’t guarantee delivery.

Listing 10-26 shows a simplified code example of Figure 10-19. The key point of interest is around the constructor of the BroadcastBlock<T>. As stated earlier, BroadcastBlock<T> can propagate the same input to many receivers. It may be important to ensure that each receiver, while getting the same data, receives them as a cloned copy. With each receiver getting their own copy this removes any potential mutable shared state issues. The BroadcastBlock<T> takes a function responsible for cloning the value to propagate; in this case you are simply using an identity function, as with a value type the value is always cloned.

Listing 10-26.  Greedy and Non-greedy Broadcast Consumers

private static void Main(string[] args)
{
   var nonGreedy = new ExecutionDataflowBlockOptions() {BoundedCapacity = 1};
   var greedy = new ExecutionDataflowBlockOptions();
 
    var source = new BroadcastBlock<int>(i => i);
 
    var consumeOne = new ActionBlock<int>((Action<int>) ConsumerOne, nonGreedy);
    var consumeTwo = new ActionBlock<int>((Action<int>) ConsumerTwo, greedy);
 
    source.LinkTo(consumeOne);
    source.LinkTo(consumeTwo);
 
    for (int j = 0; j < 10; j++)
    {
       source.Post(j);
       Thread.Sleep(50);
    }
    Console.ReadLine();
 }
 
 private static void ConsumerTwo(int obj)
 {
    Console.WriteLine("Consumer two {0}",obj);
    Thread.Sleep(60);
 }
 
 private static void ConsumerOne(int obj)
 {
   Console.WriteLine("Consumer one {0}",obj);
   Thread.Sleep(100);
 }

Running the code in Listing 10-26 produces the output shown in Figure 10-20. With "Consumer one" taking longer than "Consumer two" to process each request, and configured to be non-greedy, it does not see all requests. "Consumer two" is configured to be greedy and as such will always see all requests. Consider using a BroadcastBlock<T> when many blocks need to act upon a single message, or when live processing is more relevant than historical processing.

Joining

In situations where a block requires multiple inputs in order to perform its processing, you can utilize the JoinBlock<T1,T2> or JoinBlock<T1,T2,T3>. Consider the network described in Figure 10-21. Here we have two buffers: one that contains work items, the other network nodes capable of acting on work items. The scheduler’s job is to pair up a work item with an available node. Once it has paired up, it passes the paired value as a Tuple<T1,T2> onto the dispatcher who acts on the pair by sending the work item to the node. Once the dispatcher is aware that the node has completed the work item, the node is placed back into the work buffer. Listing 10-27 shows an implementation of such a scheduler using WCF. The key part to note is that when linking up to a JoinBlock<T1,T2>, you don’t link up directly to the block but to one of its targets. The JoinBlock then outputs a form of tuple depending on the number of blocks it has been asked to join. The implementation of the dispatcher block that actually invokes the work item is invoking the remote node asynchronously, allowing the dispatcher to immediately process any more work item node pairs. Thus there is very little need to increase the level of parallelism at this point.

Listing 10-27.  WCF-Based Grid Scheduler

[ServiceContract]
public interface IGridNode<in T>
{
  [OperationContract]
  Task InvokeAsync(T workItem);
}
 
public class GridDispatcher<T>
{
  private BufferBlock<T> workItems = new BufferBlock<T>();
  private BufferBlock<Uri> nodes = new BufferBlock<Uri>();
  private JoinBlock<Uri, T> scheduler = new JoinBlock<Uri, T>();
    
  private ActionBlock<Tuple<Uri, T>> dispatcher;
 
  public GridDispatcher(IGridNode<T> localService )
  {
    dispatcher  = new ActionBlock<Tuple<Uri, T>>((Action<Tuple<Uri, T>>) Dispatch);
 
    workItems.LinkTo(scheduler.Target2);
    nodes.LinkTo(scheduler.Target1);
 
    scheduler.LinkTo(dispatcher);
  }
 
         
  public void RegisterNode(Uri uri)
  {
    nodes.Post(uri);
  }
 
  public void SubmitWork(T workItem)
  {
    workItems.Post(workItem);
  }
 
  private void Dispatch(Tuple<Uri, T> nodeAndWorkItemPair)
  {
    var cf = new ChannelFactory<IGridNode<T>>(new NetTcpBinding(),
                        new EndpointAddress(nodeAndWorkItemPair.Item1));
 
   IGridNode<T> proxy = cf.CreateChannel();
   proxy.InvokeAsync(nodeAndWorkItemPair.Item2)
        .ContinueWith(t =>
                {
                    ((IClientChannel)proxy).Close();
                    nodes.Post(nodeAndWorkItemPair.Item1);
                });
  }
}

public class GridDispatcher<T>
{
   private BufferBlock<T> workItems = new BufferBlock<T>();
   private BufferBlock<Uri> nodes = new BufferBlock<Uri>();
   private JoinBlock<Uri, T> scheduler = new JoinBlock<Uri, T>(
                  new GroupingDataflowBlockOptions() { Greedy = false});
    
   private ActionBlock<Tuple<Uri, T>> dispatcher;
   private ActionBlock<T> localDispatcher;
 
   public GridDispatcher(IGridNode<T> localService )
   {
      localDispatcher = new ActionBlock<T>(wi =>
            {
                Console.WriteLine("Executing Locally");
                return localService.InvokeAsync(wi);
            },
            new ExecutionDataflowBlockOptions()
                    { BoundedCapacity = 2,MaxDegreeOfParallelism = 2});
 
            dispatcher  = new ActionBlock<Tuple<Uri, T>>((Action<Tuple<Uri, T>>) Dispatch);
 
            workItems.LinkTo(localDispatcher);
 
            nodes.LinkTo(scheduler.Target1);
            workItems.LinkTo(scheduler.Target2);
             
 
            scheduler.LinkTo(dispatcher);
    }
    . . .
}