Task Parallel Library and Servers, Part 3: ContinueWith

This is part 3 in a series on using Task Parallel Library when writing server applications, especially ASP.NET MVC and ASP.NET Web API applications.

1. Introduction
2. SynchronizationContext
3. ContinueWith
4. TaskHelpers

What ContinueWith Really Does

I promised that we had a bug in our previous code, and we did... sort of. Calling ContinueWith without any special flags will cause your continuation to always run, regardless of the final state of the task. We wrote our continuation under the assumption that the Task had completed successfully, which can lead to some very odd and hard to debug problems. Luckily, in our code, we ended up calling Result on the Task object, which turns around and throws an exception if the task had ended in a faulted or canceled state. But what if we'd had a Task rather than a Task<T>? Or what if we hadn't called Result? In .NET 4, this is considered a fatal error condition, and when the task object got garbage collected, its finalizer would've thrown an exception that takes down your AppDomain because you had an unobserved fault! Definitely not good.

That means that we always need to make sure to check the task's status, and if it's faulted, make sure we access the Error property of the task object so that it knows we observed the fault. That's probably a good thing to do, even if we know we're going to touch Result, because it means we get to observe the exception without causing it to be thrown again, and we definitely want to keep unnecessary exceptions to a minimum.

Unobserved Faulted Tasks in .NET 4 vs. 4.5

A quick side-note: this "killing the AppDomain because an unobserved faulted task" behavior is unique to .NET 4; it was removed from .NET 4.5. This is because all Task-based programming in .NET 4 was about explicitly returning a Task object. When the team added the async and await keywords to .NET 4.5, they also enabled async methods to return void. They discovered that there was a very useful place for void-returning async methods: as event handlers in UI applications. Since the method returns void, there's no way for the caller to know (or care) that the method they called actually spun off async work to be done; as such, it could no longer be the case that unobserved faults should be an application failure.

That said, these event handlers in client UI applications may be good candidates for void-returning async methods; however, you must never use such a method in server event handling (for example, when handling a WebForms event) when the client needs to wait for the result. The event handler that calls your callback has no way of knowing that you have some async work to be done and that it should wait for you to finish, so of course the client will get its response before your work is complete.

Performance Optimizations

One more thing we can do with our continuation work is to avoid calling ContinueWith at all. This has two potential benefits:

1. We can avoid the potential thread hop that comes with ContinueWith;
2. We can avoid allocating an extra Task object when not needed.

Both of these reasons are good reasons why we should consider helpers for ContinueWith when running on servers.

Revising Our ContinueWith Example

Let's start with our example from part 2:

var ctxt = SynchronizationContext.Current;

return SomeThingReturningATask()
      .ContinueWith(innerTask =>
{
    var tcs = new TaskCompletionSource<int>();

    try
    {
        if (ctxt != null)
        {
            ctxt.Post(state =>
            {
                try
                {
                    int result = innerTask.Result;
                    tcs.TrySetResult(result);
                }
                catch(Exception ex)
                {
                    tcs.TrySetException(ex);
                }
            }, state: null);
        }
        else
        {
            int result = innerTask.Result;
            tcs.TrySetResult(result);
        }
    }
    catch(Exception ex)
    {
        tcs.TrySetException(ex);
    }

    return tcs.Task;
});

Let's extract a helper function out of this:

static Task<TOut> Continue(
        this Task<TIn> task,
        Func<Task<TIn>, TOut> next)
{
    var tcs = new TaskCompletionSource<TOut>();
    var ctxt = SynchronizationContext.Current;

    task.ContinueWith(innerTask =>
    {
        try
        {
            if (ctxt != null)
            {
                ctxt.Post(state =>
                {
                    try
                    {
                        var res = next(innerTask);
                        tcs.TrySetResult(res);
                    }
                    catch(Exception ex)
                    {
                        tcs.TrySetException(ex);
                    }
                }, state: null);
            }
            else
            {
                var res = next(innerTask);
                tcs.TrySetResult(res);
            }
        }
        catch(Exception ex)
        {
            tcs.TrySetException(ex);
        }
    });

    return tcs.Task;
}

This is the same code as we had above, but now we just call this with a very simple lambda, and all the work is done on our behalf:

return SomeThingReturningATask()
      .Continue(innerTask => innerTask.Result * 42);

The helper is taking care of the transition to the sync context automatically for us. Now as we add new capabilities to this helper function, we reap the benefits everywhere we're using it.

Removing ContinueWith

The next optimization we want to do is remove the call to ContinueWith entirely if the task we've been given is already completed. This prevents us from hopping onto a new thread to get the work done. In a stack like Web API which has been designed from top to bottom to be async, it turns out that most everything we do completes synchronously, because on the server there are very few reasons to actually wait (remember, we know that we shouldn't thread-hop just to compute something, so our opportunities for async on a server are usually around long-running I/O operations).

If we re-work our example to avoid unnecessary calls to ContinueWith, then it might look something like this:

static Task<TOut> Continue(
        this Task<TIn> task,
        Func<Task<TIn>, TOut> next)
{
    var tcs = new TaskCompletionSource<TOut>();
    var ctxt = SynchronizationContext.Current;

    if (task.IsCompleted)
    {
        try
        {
            var res = next(innerTask);
            tcs.TrySetResult(res);
        }
        catch(Exception ex)
        {
            tcs.TrySetException(ex);
        }
    }
    else
    {
        task.ContinueWith(innerTask =>
        {
            try
            {
                if (ctxt != null)
                {
                    ctxt.Post(state =>
                    {
                        try
                        {
                            var res = next(innerTask);
                            tcs.TrySetResult(res);
                        }
                        catch(Exception ex)
                        {
                            tcs.TrySetException(ex);
                        }
                    }, state: null);
                }
                else
                {
                    var res = next(innerTask);
                    tcs.TrySetResult(res);
                }
            }
            catch(Exception ex)
            {
                tcs.TrySetException(ex);
            }
        });
    }

    return tcs.Task;
}

In our task.IsCompleted block, notice that I don't bother with the sync context. That's because we're staying on the same thread, so the sync context will be the same. We also have three potential threads of execution in here (the main function body, plus the two lambdas) so we have three try/catch handlers, converting exceptions into faulted tasks.

Lambdas and Closures

There is one final bit of optimization left to be done. Now that we have a block of code which calls ContinueWith and a block of code that doesn't, we should be paying close attention to whether our lambda functions are actually closures.

A closure is a special kind of lambda that has access to the local variables and parameters of the function that it's inside of, even after that function has exited. If you look closely at our call to ContinueWith, you'll see that we are in fact writing a closure, because we use the next parameter as well as the tcs and ctxt variables inside the lambda. If we were diligent, we could get rid of tcs, but we're stuck with the others.

When the compiler generates a closure, it does so by making a tiny object with the state inside of it. Unfortunately, when the compiler realizes that you've got a closure anywhere in your method, it does all the work up front of allocating the state object, even if you never end up using the closure (and if you have two closures, that state object ends up being a combination of the shared values between both closures). When you see code like this which is only sometimes using a closure, it's common to split it up into the non-closure part vs. the closure part.

Let's iterate one more time on the helper function:

static Task<TOut> Continue(
        this Task<TIn> task,
        Func<Task<TIn>, TOut> next)
{
    if (task.IsCompleted)
    {
        var tcs = new TaskCompletionSource<TOut>();

        try
        {
            var res = next(innerTask);
            tcs.TrySetResult(res);
        }
        catch(Exception ex)
        {
            tcs.TrySetException(ex);
        }

        return tcs.Task;
    }

    return ContinueClosure(task, next);
}

static Task<TOut> ContinueClosure(
        Task<TIn> task,
        Func<Task<TIn>, TOut> next)
{
    var ctxt = SynchronizationContext.Current;

    return task.ContinueWith(innerTask =>
    {
        var tcs = new TaskCompletionSource<TOut>();

        try
        {
            if (ctxt != null)
            {
                ctxt.Post(state =>
                {
                    try
                    {
                        var res = next(innerTask);
                        tcs.TrySetResult(res);
                    }
                    catch(Exception ex)
                    {
                        tcs.TrySetException(ex);
                    }
                }, state: null);
            }
            else
            {
                var res = next(innerTask);
                tcs.TrySetResult(res);
            }
        }
        catch(Exception ex)
        {
            tcs.TrySetException(ex);
        }

        return tcs.Task;
    }).Unwrap();
}

We've split the work into two functions, and we've moved the allocation of the TaskCompletionSource inside the closure. We still have next and ctxt in the closure, but there's nothing we can do about that since there's no way to pass a 'state' object into ContinueWith in .NET 4. Since we moved the TCS into closure, now the ContinueWith call ends up returning a Task<Task<TOut>>, so we use the Unwrap extension method, whose job is to shed that outer Task.

Unobserved Faulted Tasks

I started this blog post talking about the bug with unobserved faulted tasks, but... I didn't actually fix that problem. We still need to be sure to observe innerTask.Result so that any faulted task can re-throw its exception appropriately, and we won't end up with a task whose errors weren't observed bringing down our AppDomain. In addition, it seems silly for us to throw exceptions by observing a faulted task just so that it can be turned back into another faulted task, so we'll see if we can deal with that, too.

In the next post in the series, I'll talk about splitting out this one helper method into several optimized methods for each specific scenario (successful results, faulted results, and cancelation), which will also cure of us needing to observe properties on the Task object just to ensure we don't crash later.

Comments

[Bilal]

Hi Brad,

I am using TPL to run a stored procedure and update records in various tables. The stored procedure is slow and for a single entity takes about 30 secs.
When I run it using Task everything seems to work fine but the records sometimes update and sometimes do not.

Any ideas, pointers are much appreciated.
I posed the question on stackoverflow http://stackoverflow.com/questions/10121876/calling-stored-procedure-inside-a-thread-to-update-multiple-records

[Brad Wilson]

Well, I see several problems.

1. Your tasks aren't actually doing asynchronous I/O, so they're blocking all the threads that they've been assigned, and the thread pool is a shared resource with ASP.NET so you're actually taking away threads that the server wants to use to process requests.

2. You contend there are no locks, but any database which is doing something for 30 seconds must be locked somewhere. Even inserts can lock a table if they're being done in a transaction and the primary key is an auto-number field.

3. Your call to Task.WaitAll() blocks, stealing yet another thread from ASP.NET that it would prefer to be using to service requests.

[vetal01]

Hi Brad. There are some bugs in your code

static Task<TOut> Continue<TIn, TOut>(
this Task<TIn> task,
Func<Task<TIn>, TOut> next)
{
if (task.IsCompleted)
{
var tcs = new TaskCompletionSource<TOut>();

try
{
var res = next(task);
tcs.TrySetResult(res);
}
catch (Exception ex)
{
tcs.TrySetException(ex);
}

return tcs.Task;
}

return ContinueClosure(task, next);
}

static Task<TOut> ContinueClosure<TIn, TOut>(
Task<In> task,
Func<Task<TIn>, TOut> next)
{
var ctxt = SynchronizationContext.Current;

return task.ContinueWith(innerTask =>
{
var tcs = new TaskCompletionSource<TOut>();

try
{
if (ctxt != null)
{
ctxt.Post(state =>
{
try
{
var res = next(innerTask);
tcs.TrySetResult(res);
}
catch (Exception ex)
{
tcs.TrySetException(ex);
}
}, state: null);
}
else
{
var res = next(innerTask);
tcs.TrySetResult(res);
}
}
catch (Exception ex)
{
tcs.TrySetException(ex);
}

return tcs.Task;
})
.Unwrap();
}

[Bilal]

So, What would you suggest to make it asynchronous?

[noname]

Ur great at writing articles that are not accessible to the average programmer, perhaps you should of mentioned that knowledge about the task library and the new async features were mandatory in order to be able to interpret and understand your article.

[Omegaluz.wordpress.com]

Brad,

You said "That means that we always need to make sure to check the task's status, and if it's faulted, make sure we access the Error property of the task object so that it knows we observed the fault. That's probably a good thing to do, even if we know we're going to touch Result, because it means we get to observe the exception without causing it to be thrown again, and we definitely want to keep unnecessary exceptions to a minimum."

I don't see an "Error" property in the .Net 4 Task object. Don't you mean the "Exception" property?

[Micahsays]

Hi brad-- any updates on Task testing (and xUnit)? So far i've just been waiting on completion. Id like to see an example of a test you use just so that i can see your method, if possible.

Thanks for writing this series-- i've been finding myself on these pages fairly frequently recently.

[Brad Wilson]

With 1.9 and later, you can use async Task-returning test methods, and xUnit.net will wait for the Task to complete automatically. For example:

[Fact]
public async Task MyTest()
{

    var result = await SomeAsyncMethod();
    Assert.Equal(42, result);
}

If you're still using .NET 4, you can also use ContinueWith, though much of the syntactic value is lost:

[Fact]
public Task MyTest()
{
    return SomeAsyncMethod().ContinueWith(t => {
        var result = t.Result;
        Assert.Equal(42, result);

    }
}

[Micahsays]

Wow-- thanks for getting back to me so quick. I especially appreciate showing me the 4.0 one as that's what i'm on (because i'm on azure).

You know i had that in my code-- but i obviously confused NCrunch not showing the final assert status as meaning that TPL Async wasn't xUnit compatible.

Thanks again.