Bnaya Eshet : async, Parallel, .NET 4.5

Async - Handling multiple Exceptions

bnaya — Mon, 10 Dec 2012 06:30:25 GMT

Async - Handling multiple Exceptions

the SDP conference was ended a few week ago and I finally find a time to write some comments.

It was a very successful conference, the feedback and evaluations, of most sessions scored higher than 4.5, some well-over it. For example, one workshop had a perfect 5 / 5 score, and two other workshops scored 4.92 / 5 and 4.9 / 5. The highest score for a breakout session was 9.39 / 10, which is the highest score we’ve seen to date. my score was 4.90 / 5.

but this post isn't about scores, it is about fixing a sample that I was showing at the SDP about exception handling within the async / await context.

when you do use the great async / await syntax you may run into a problem while trying to handle multiple exception which was thrown from async operations (it can happens in cases like parent child relationship or when using WhenAll API).

by design the catch closure will handle the first exception and ignore all the other, you can read about it in here.

the following code will catch a single NullReferenceException or ArgumentException exception (the AggregateException will be ignored):

Code Snippet
     try
    {
        await Task.Factory.StartNew(() =>
            {
                Task.Factory.StartNew(() => 
                    { throw new NullReferenceException(); },
                        TaskCreationOptions.AttachedToParent);
                Task.Factory.StartNew(() => 
                    { throw new ArgumentException(); },
                        TaskCreationOptions.AttachedToParent);
            });
    }
    catch (AggregateException ex)
    {
        // this catch will never be target
    }
    catch (Exception ex)
    {
        Console.WriteLine("## {0} ##", ex.GetType().Name);
    }
}
 

if you do want to catch the AggregateException you can use the following work around :

Code Snippet
 try
{
    Task t = Task.Factory.StartNew(() =>
        {
            Task.Factory.StartNew(() => 
                { throw new NullReferenceException(); },
                    TaskCreationOptions.AttachedToParent);
            Task.Factory.StartNew(() => 
                { throw new ArgumentException(); },
                    TaskCreationOptions.AttachedToParent);
        });
    await t.ContinueWith(tsk => { if (tsk.Exception != null) throw tsk.Exception; });
}
catch (AggregateException ex)
{
    foreach (var exc in ex.Flatten().InnerExceptions)
    {
        Console.WriteLine(exc.GetType().Name);
    }
}
 

the difference is at line 12, I'm using a continuation that re-throw the AggregateException.

because this is a fairly common practice you may better wrap it within an extension method like the following one:

Code Snippet
 public static Task ThrowMultiple(this Task t)
{
    return t.ContinueWith(tsk => { if (tsk.Exception != null) throw tsk.Exception; });
}
 
public static Task<T> ThrowMultiple<T>(this Task<T> t)
{
    return t.ContinueWith(tsk => 
    {
        if (tsk.Exception != null)
            throw tsk.Exception;
        return tsk.Result;
    });
}
 

you can use it as follow:

Code Snippet
 try
{
    await Task.Factory.StartNew(() =>
        {
            Task.Factory.StartNew(() =>
                { throw new NullReferenceException(); },
                    TaskCreationOptions.AttachedToParent);
            Task.Factory.StartNew(() =>
                { throw new ArgumentException(); },
                    TaskCreationOptions.AttachedToParent);
        }).ThrowMultiple();
}
catch (AggregateException ex)
{
    foreach (var exc in ex.Flatten().InnerExceptions)
    {
        Console.WriteLine(exc.GetType().Name);
    }
}
 

Summary

catching a single exception within an async method is a design decision,
but you may want to override this decision in some scenarios.

I hope that in a future release of .NET framework Microsoft consider to throw an AggregateException when the compiler will detect that the developer was adding an AggregateException catch block.

in the meantime you can use the extension methods.

you should not forget to catch a non AggregateException if you have any code before the await (this code will run synchronously).

SDP 2012 - Day 1

bnaya — Sun, 18 Nov 2012 21:21:40 GMT

SDP 2012 - Day 1

the SDP 2012 conference has started today.

I was speaking about what's new in .NET 4.5 parallelism.

you can find the materials for my lecture in here.

tomorrow I'm giving a full day tutorial about async / await, Rx and TPL Dataflow.

async / await, some reasoning

bnaya — Thu, 19 Jan 2012 16:09:23 GMT

async / await, some reasoning

this post will try to make some reasoning about the .NET 4.5 / C#5 await keyword.

I will begin with a quiz.

how long will it take to the following method to produce the 42 value?

Code Snippet
 async Task<int> Execute()
{
    await Task.Delay(1000);
    await Task.Delay(1000);
    return 42;
}
 

you should remember that conceptually the await keyword will translate to a continuation.

the above code can be compare to the following TPL 4 code snippet:

Code Snippet
 Task<int> Execute()
{
    Task t1 = Task.Factory.StartNew(
        () => Thread.Sleep(1000));
    Task t2 = t1.ContinueWith(t1_ =>
        {
            Thread.Sleep(1000);
        });
    Task<int> t3 = t2.ContinueWith(t2_ => 42);
    return t3;
}
 

whatever come after the await will be compile into a continuation closure.

therefore the 42 value will be produce after 2 second.

How can we await for multiple task?

there is couple of way for awaiting on multiple tasks, but the most recommended one is to use Task.WhenAll (you can also use Task.WhenAny to continue after the completion of the first task).

do not confuse the Task.WhenAll with Task.WaitAll, WhenAll is a continuation which happens when all the task come to completion, while WaitAll is a blocking API which will block the execution until all tasks will be completed.

the following snippet demonstrate the Task.WhenAll usage.

Code Snippet
 async Task<int> Execute()
{
    Task t1 = Task.Delay(1000);
    Task t2 = Task.Delay(1000);
    await Task.WhenAll(t1, t2);
    return 42;
}
 

the 42 value will now produce after 1 second.

it is somewhat equals to the following TPL 4 snippet:

Code Snippet
 Task<int> Execute()
{
    Task t1 = Task.Factory.StartNew(
        () => Thread.Sleep(1000));
    Task t2 = Task.Factory.StartNew(
        () => Thread.Sleep(1000));
    Task<int> t3 = Task.Factory.ContinueWhenAll(
        new[]{t1, t2}, tsks => 42);
    return t3;
}
 

to wrap it up let think how long will it take to the following snippet to produce a value.

Code Snippet
 async Task<int> Execute()
{
    for (int i = 0; i < 10; i++)
    {
        await Task.Delay(1000);
    }
    return 42;
}
 

the right answer is 10 seconds, each iteration will result in a continuation closure which will wrap the following iterations.

this can be translate to something like the following snippet (TPL 4).

Code Snippet
 Task<int> Execute()
{
    var stateMachine = new StateMachine();
    return stateMachine.OnNext();
}
 
class StateMachine
{
    private int _i;
    private TaskCompletionSource<int> _semanticTask =
        new TaskCompletionSource<int>();
 
    public Task<int> OnNext()
    {
        Interlocked.Increment(ref _i);
 
        Task t = Task.Factory.StartNew(() =>
            Thread.Sleep(1000));
        if (_i <= 10)
            t.ContinueWith(t_ => OnNext());
        else
            _semanticTask.SetResult(42);
 
        return _semanticTask.Task;
    }
}
 

the execute method will create a state machine which will chain task continuation 10 times and then set the value of 42.
the TaskCompletionSource represent semantics of task (TAP - Task Async Pattern).

TaskCompletionSource does not produce any concurrency (doesn't attached to any thread), it just present a task which can be project result, exception or cancellation.

the OnNext method immediately return a semantic task which will signal as complete at line 22 (the exit term of the recursion).

Summary

await present a continuation. each time the code is hitting the await it construct a new continuation closure.

Using async / await

bnaya — Sun, 15 Jan 2012 05:46:58 GMT

Using async / await

this post will discuss parallel disposal.

whenever we want to dispose a parallel execution upon completion we can't use the convenient using keyword.

for example, the following code may be dispose the command before completion:

Very bad Code Snippet
 using (var conn = new SqlConnection(CONN_STR))
using (var cmd = new SqlCommand("Select * from Employee", conn)) 
{
    conn.Open();
    cmd.BeginExecuteReader(ar =>
        {
            int affected = cmd.EndExecuteNonQuery(ar);
        });
}
 

the using is absolutely wrong for the above sample.

what should be done is:

Code Snippet
 var conn = new SqlConnection(CONN_STR);
var cmd = new SqlCommand("Select * from Employee", conn);
conn.Open();
cmd.BeginExecuteReader(ar =>
{
    int affected = cmd.EndExecuteNonQuery(ar);
    cmd.Dispose();
    conn.Dispose();
}, null);
 

we can write it slightly friendlier by using the TPL FromAsync wrapper:

Code Snippet
 var conn = new SqlConnection(CONN_STR);
var cmd = new SqlCommand("Select * from Employee", conn);
conn.Open();
Task<int> t = Task.Factory.FromAsync<int>(
    cmd.BeginExecuteReader,
    cmd.EndExecuteNonQuery,
    null);
t.ContinueWith(tsk =>
    {
        int affected = tsk.Result;
        cmd.Dispose();
        conn.Dispose();
    });
 

but what if the compiler can rewrite our code?

in that case we can write a similar code to the code in the first code snippet and get it compiled into something like the above code snippet above.

this is exactly what happens when in .NET 4.5 / C#5 (async / await pattern).

when we write the following code:

Code Snippet
 async Task<int> ExecuteNonQueryAsync()
{
    int affected = 0;
    using (var conn = new SqlConnection(CONN_STR))
    using (var cmd = new SqlCommand("Select * from Employee", conn))
    {
        conn.Open();
        affected = await Task.Factory.FromAsync<int>(
            cmd.BeginExecuteReader,
            cmd.EndExecuteNonQuery,
            null);
    }
    return affected;  
}
 

the compiler does rewrite this async method.

everything that follow the await keyword (line 8),

will be put into a continuation state machine, including the closing of the curly brackets (of the using).

Summary

using the new async / await pattern will dispose our resources on time.

using keyword is a very clean syntax and now we can apply it to parallel execution.

async \ await and Exception Handling

bnaya — Thu, 12 Jan 2012 13:11:07 GMT

async \ await and Exception Handling

this post will discuss how async / await is handling exceptions.

as we mention in previous post, about the async / await concept, await is all about continuation.

before .NET 4.5 parallel execution exceptions has to be handle in separate of the synchronic handling.

for example:

handling ThreadPool execution:

Code Snippet
 void Foo()
{
    try
    {
        Console.WriteLine("Synchronic");
        ThreadPool.QueueUserWorkItem(state =>
            {
                try
                {
                    Console.WriteLine("Parallel");
                }
                catch (Exception exAsync)
                {
                    EventLog.WriteEntry("application", exAsync.ToString());
                }
            }, null);
    }
    catch (Exception ex)
    {
        EventLog.WriteEntry("application", ex.ToString());
    }
}
 

as you can see we have to handle the parallel exception (line 12) in separate from the synchronic handling (line 18).

TPL has brought new option for handling parallel exception, now you can use ContinueWith with TaskContinuationOptions.OnlyOnFault (line 8).
but still you have to handle the parallel exception is in separate of the synchronic one:

Code Snippet
 void Foo()
{
    try
    {
        Console.WriteLine("Synchronic");
        Task t = Task.Factory.StartNew(() => Console.WriteLine("Parallel"));
        t.ContinueWith(tsk => EventLog.WriteEntry("application", tsk.Exception.ToString()), 
            TaskContinuationOptions.OnlyOnFaulted);
    }
    catch (Exception ex)
    {
        EventLog.WriteEntry("application", ex.ToString());
    }
}
 

async / await pattern

using the async / await pattern we can handle both synchronic and parallel exception in the same place:

Code Snippet
 async void Foo()
{
    try
    {
        Console.WriteLine("Synchronic");
        await Task.Factory.StartNew(() => Console.WriteLine("Parallel"));
    }
    catch (Exception ex)
    {
        // handling both synchronic and parallel exceptions
        EventLog.WriteEntry("application", ex.ToString());
    }
}
 

when we are using the async / await pattern at compile time the compiler convert our code into continuation state machine.
therefore the compiler can take the code within the catch area and apply it both for the synchronic and the parallel execution.

Summary

async / await pattern does simplify the exception handling. we do write our exception handling once and it will apply for both synchronic and parallel execution.

the concept of async \ await

bnaya — Thu, 29 Dec 2011 22:41:42 GMT

the concept of async \ await

in this post I will survey the new .NET 4.5 / C# 5 concept of async / await.

I will focus on how to understand what is really happens behind the new async / await syntax.

What's it all about?

the new async / await syntax is using the C# syntactic compiler to generate async operation from code that is looking very much like a synchronous code.

but before we start we should discus the new C# 5 syntax.

the syntax include 2 keywords:

async - which is only a marker for async method.
await - indicate a callback boundary.

Code Snippet
 static async Task Execute()
{
    Console.WriteLine(run on calling thread);
 
    await Task.Factory.StartNew(() => Thread.Sleep(1000));
 
      Console.WriteLine(run on callback thread); 
}
 

so how should we understand what was written in the above code?

actually it is a different way to represent a continuation (you can read more about the continuation concept in here).

the above code is somewhat identical to the following TPL 4 code:

Code Snippet
 static Task Execute()
{
    Console.WriteLine(run on calling thread);
    Task t = Task.Factory.StartNew(() => Thread.Sleep(1000));
    return t.ContinueWith (tsk =>
        {
              Console.WriteLine(run on callback thread);
        });
}
 

the syntactic compiler will translate the code below the await keyword into continuation state machine, which is logically (not technically) identical to the above code.

Point of interest:

you may have been notice that the async method return a Task even though there is no return within the method block.
surveying the TPL 4 code snippet we can understand that the async method will actually return to the caller immediately after the Task.Factory.StartNew start the task and the rest of the code is actually a continuation callback.

what we got back from the async method is a task which represent the async part of the method.

async / await with return value

async / await can represent a continuation of a callback that accept async result.

Code Snippet
 static async Task<DateTime> Execute()
{
    DateTime result = await Task.Factory.StartNew(() => DateTime.Now );
 
      return result.AddDays(1);
}
 

the above code will logically translate to:

Code Snippet
 static Task<DateTime> Execute()
{
    Task<DateTime> t = Task.Factory.StartNew(() => DateTime.Now);
    return t.ContinueWith (tsk =>
        {
              return tsk.Result.AddDays(1);
        });
}
 

you may have notice that the return value (on the left side of the await) was unwrapped (DateTime instead of Task<DateTime>)

Which thread is running?

normally when the method doesn't invoke from the UI thread, everything before the await line will run synchronously on the caller thread.
the Task.Run naturally will be schedule on a different thread and everything under the await will be schedule on different thread then the caller thread, it may be the same thread of the Task.Run or any other ThreadPool thread (when there is only single continuation it will probably be the same thread as Task.Run)

Async and UI

whenever the async method invocation is coming from UI thread (or to be more precise from thread under synchronization context) the continuation return back to the synchronization context thread.

this is quit similar to the following TPL code (.NET 4):

Code Snippet
 TaskScheduler scheduler = TaskScheduler.FromCurrentSynchronizationContext();
Task t = Task.Factory.StartNew(() => Trace.WriteLine("in parallel"));
t.ContinueWith(tsk => Trace.WriteLine("UI thread"), scheduler);
 

or to more legacy code which is using the synchronization context directly:

Code Snippet
 Action a = () => Trace.WriteLine("in parallel");
SynchronizationContext sc = SynchronizationContext.Current;
a.BeginInvoke(ar => 
    {
        sc.Post(state => Trace.WriteLine("UI thread"), null);
    }, null);
 

async / await is aware of the synchronization context of the caller and if any it schedule the await callback on this context.

Summary

the syntactic compiler translate the async / await syntax into state machine which handle the continuation flow after parallel operation.

there is much more for that and I will discuss it in future posts.

you can see it performance characteristic on this post.