Exception Handling In DataBlocks

Question

I am trying to understand exception handling in TPL.

The following code seems to swallow exceptions:

var processor = new ActionBlock<int>((id) => SomeAction(id), new ExecutionDataflowBlockOptions { ... });


async Task SomeAction(int merchantID)
{
    //Exception producing code
    ...
}

And listening to TaskScheduler.UnobservedTaskException events does not receive anything either.

So, does this mean the action block does a try-catch in itself when running the actions?

Is there any official documentation of this somewhere?

Answer 1

The TaskScheduler.UnobservedTaskException event is not a reliable/deterministic way to handle exceptions of faulted tasks, because it's delayed until the faulted task is cleaned up by the garbage collector. And this may happen long after the error occurred.

The only type of exception that is swallowed by the dataflow blocks is the OperationCanceledException (by design). All other exceptions result to the block transitioning to a faulted state. A faulted block has its Completion property (which is a Task) faulted as well (processor.Completion.IsFaulted == true). You can attach a continuation to the Completion property, to receive a notification when a block fails. For example you could ensure that an exception will not pass unnoticed, by simply crashing the process:

processor.Completion.ContinueWith(t =>
{
    ThreadPool.QueueUserWorkItem(_ => throw t.Exception);
}, default, TaskContinuationOptions.OnlyOnFaulted, TaskScheduler.Default);

This works because throwing an unhandled exception on the ThreadPool causes the application to terminate (after raising the AppDomain.CurrentDomain.UnhandledException event).

If your application has a GUI (WinForms/WPF etc), then you could throw the exception on the UI thread, that allows more graceful error handling:

var uiContext = SynchronizationContext.Current;
processor.Completion.ContinueWith(t =>
{
    uiContext.Post(_ => throw t.Exception, null);
}, default, TaskContinuationOptions.OnlyOnFaulted, TaskScheduler.Default);

This will raise the Application.ThreadException event in WinForms.

Answer 2

Update

The exception handling behavior of DataFlow blocks is explained in Exception Handling in TPL DataFlow Networks

**Original

This code doesn't swallow exceptions. If you await the block to complete with await processor.Completion you'll get the exception. If you use a loop to pump messages to the block before calling Complete() you need a way to stop the loop too. One way to do it is to use a CancellationTokenSource and signal it in case of exception:

void SomeAction(int i,CancellationTokenSource cts)
{
    try
    {
        ...
    }
    catch(Exception exc)
    {
        //Log the error then
        cts.Cancel();
       //Optionally throw
    }
}

The posting code doesn't have to change all that much, it only needs to check whether

var cts=new CancellationTokenSource();
var token=cts.Token;
var dopOptions=new new ExecutionDataflowBlockOptions { 
                           MaxDegreeOfParallelism=10,
                           CancellationToken=token
};
var block= new ActioBlock<int>(i=>SomeAction(i,cts),dopOptions);

while(!token.IsCancellationRequested && someCondition)
{
    block.Post(...);
}
block.Complete();
await block.Completion;

When the action throws, the token is signaled and the block ends. If the exception is rethrown by the action, it will be rethrown by await block.Completion as well.

If that seems convoluted, it's because that's somewhat of an edge case for blocks. DataFlow is used to create pipelines or networks of blocks.

The general case

The name Dataflow is significant. Instead of building a program by using methods that call each other, you have processing blocks that pass messages to each other. There's no parent method to receive results and exceptions. The pipeline of blocks remains active to receive and process messages indefinitely, until some external controller tells it to stop, eg by calling Complete on the head block, or signaling the CancellationToken passed to each block.

A block shouldn't allow unhandled exceptions to occur, even if it's a standalone ActionBlock. As you saw, unless you've already called Complete() and await Completion, you won't get the exception.

When an unhandled exception occurs inside a block, the block enters the faulted state. That state propagates to all downstream blocks that are linked with the PropagateCompletion option. Upstream blocks aren't affected, which means they may keep working, storing messages in their output buffers until the process runs out of memory, or deadlocks because it receives no responses from the blocks.

Proper Failure Handling

The block should catch exceptions and decide what to do, based on the application's logic:

1. Log it and keep processing. That's not that different from how web application's work - an exception during a request doesn't bring down the server.
2. Send an error message to another block, explicitly. This works but this type of hard-coding isn't very dataflow-ish.
3. Use message types with some kind of error indicator. Perhaps a Success flag, perhaps an Envelope<TMessage> object that contains either a message or an error.
4. Gracefully cancel the entire pipeline, by signaling all blocks to cancel by signaling the CancellationTokenSource used to produce the CancellationTokens used by all blocks. That's the equivalent of throw in a common program.

#3 is the most versatile option. Downstream blocks can inspect the Envelope and ignore or propagate failed messages without processing. Essentially, failed messages bypass downstream blocks.

Another option is to use the predicate parameter in LinkTo and send failed messages to a logger block and successful messages to the next downstream block. In complex scenarios, this could be used to eg retry some operations and send the result downstream.

These concepts, and the image, come from Scott Wlaschin's Railway Oriented Programming