Parallel Programming with .NET

Potential pitfalls to avoid when passing around async lambdas

Stephen Toub - MSFT — Wed, 08 Feb 2012 17:02:24 GMT

One of the really useful capabilities of the new async methods feature in C# and Visual Basic is the ability to write async lambdas and anonymous methods (from here on in this post, I’ll refer to both of these as async lambdas, since the discussion applies equally to both). This allows you to easily get a delegate to represent an asynchronous operation, e.g.

Func<Uri,Task<string>> getContentsLowerCaseAsync = async url =>
{
string contents = await DownloadString(url);
return contents.ToLower();
};

Async methods in C# and Visual Basic can return void, Task, or Task<TResult>, which means they can be mapped to delegates that return void, Task, or Task<TResult>. This is very powerful, but it can also lead to subtle bugs if you’re not careful.

Most methods today that accept as a parameter a delegate that returns void (e.g. Action, Action<T>, etc.) expect the work of that delegate to be completed by the time the delegate completes. As a simple example, consider a timing helper function, whose job it is to time how long a particular piece of code takes to execute:

public static double Time(Action action, int iters=10)
{
    var sw = Stopwatch.StartNew();
    for(int i=0; i<iters; i++) action();
    return sw.Elapsed.TotalSeconds / iters;
}

With this function, if I then run the following code:

static void Main()
{
    double secs = Time(() =>
    {
        Thread.Sleep(1000);
    });
    Console.WriteLine("Seconds: {0:F7}", secs);
}

I see this written out to the console:

Seconds: 0.9999956
Press any key to continue . . .

That’s what I’d expect: we asked to sleep for one second, and that’s almost exactly what the timing showed. But now consider an alternate piece of code:

static void Main()
{
    double secs = Time(async () =>
    {
        await Task.Delay(1000);
    });
    Console.WriteLine("Seconds: {0:F7}", secs);
}

When I run this, I see the following written out to the console:

Seconds: 0.0000341
Press any key to continue . . .

Huh? Here we have an async method that’s awaiting a Task that won’t complete for a second, so this asynchronous method’s execution should also be at least a second, and yet the timer is telling us that it took only 34 microseconds? What’s going on?

To understand this effect, we need to remember how async methods operate. When you invoke an async method, it starts running synchronously. If the method doesn’t have any awaits in it, or if all of the awaits in the method are on awaitables that are already completed by the time they’re awaited, then the method will run entirely synchronously. However, when the method encounters the first await that yields, the async method returns. In the case of an async method that returns a Task or a Task<TResult>, the method at this point returns the Task or Task<TResult> that represents the async method’s execution, and the caller can use that task to wait synchronous (e.g. Wait()) or asynchronously (e.g. await, ContinueWith) for the method to asynchronously complete. In the case of a void method, though, no handle is handed back. Async void methods are thus often referred to as “fire and forget.”

Now with that background, consider what’s happening with our timing function. Our Time method accepts an Action, so the compiler is going to map our “async () => { … }” to being a void-returning async method, and the Action passed into the Time method will be for that void method. Thus, when Time invokes the Action, the Action will return as soon as it hits the first await that yields, which is our await for the delay task. This means that we’re really only timing the invocation of the async method up until the await, but not including the time to await the task or what comes after it.

We can fix this by modifying our Time function to accept a Func<Task> instead of an Action:

public static double Time(Func<Task> func, int iters=10)
{
    var sw = Stopwatch.StartNew();
    for (int i = 0; i < iters; i++) func().Wait();
    return sw.Elapsed.TotalSeconds / iters;
}

Now when I compile and run our async lambda, I get the following output that’s what I’d expect:

Seconds: 1.0078671
Press any key to continue . . .

Void-returning methods aren’t the only potentially problematic area; they’re just the easiest example to highlight, because it’s very clear from the signature that they don’t return anything and thus are only useful for their side-effects, which means that code invoking them typically needs them to run to completion before making forward progress (since it likely depends on those side-effects having taken place), and async void methods defy that.

A more complicated but still problematic example is a generic method that accepts an Action as a parameter and returns a Task, or that accepts a Func<…,TResult> as a parameter and returns a Task<TResult>, such as Task.Factory.StartNew. Consider the following:

var t = Task.Factory.StartNew(() =>
{
Thread.Sleep(1000);
return 42;
});

Here StartNew accepts a delegate of type Func<int>, and returns a Task<int> representing the execution of the Func<int> delegate. Makes sense. But now consider the following:

var t = Task.Factory.StartNew(async () =>
{
await Task.Delay(1000);
return 42;
});

Any guesses as to what the type of ‘t’ is? StartNew accepts a Func<TResult> and returns a Task<TResult>. We’re passing in an async lambda that will give back a Task<int>, which means the TResult in Func<TResult> is actually Task<int>, such that the delegate provided to StartNew is a Func<Task<int>>. That means that this call to StartNew is actually returning a Task<Task<int>>. The task created by StartNew will invoke the Func<Task<int>>, which will run synchronously until the first await that yields, at which point the Func<Task<int>> will return, handing back the result Task<int> that represents the async lambda’s execution. StartNew will then complete the Task<Task<int>> that it handed back, since the delegate associated with that task has completed its synchronous execution. If I wrote code that depended on the returned task’s completion to mean that the async lambda had completed, I’d be sorely disappointed. It’s actually the returned task’s Result (which is itself a Task<int>) that represents the async lambda.

There are a few ways to address this, such as using the Unwrap method:

var t = Task.Factory.StartNew(async () =>
{
await Task.Delay(1000);
return 42;
}).Unwrap();

For more information, see my previous blog post on this (and on how Task.Run differs in behavior here from Task.Factory.StartNew) at http://blogs.msdn.com/b/pfxteam/archive/2011/10/24/10229468.aspx.

When “ExecuteSynchronously” doesn’t execute synchronously

Stephen Toub - MSFT — Tue, 07 Feb 2012 19:22:47 GMT

When creating a task continuation with ContinueWith, developers have the opportunity to provide a TaskContinuationOptions enum value, which could include the TaskContinuationOptions.ExecuteSynchronously flag. ExecuteSynchronously is a request for an optimization to run the continuation task on the same thread that completed the antecedent task off of which we continued, in effect running the continuation as part of the antecedent’s transition to a final state (the default behavior for ContinueWith if this flag isn’t specified is to run the continuation asynchronously, meaning that when the antecedent task completes, the continuation task will be queued rather than executed). I explicitly used the word “request” in the previous sentence, because while TPL strives to honor this as much as possible, there are in fact a few cases where continuations created with TaskContinuationOptions.ExecuteSynchronously will still run asynchronously. (What follows is a discussion of TPL’s current implementation in .NET 4 and the .NET 4.5 Developer Preview.)

One condition that will force a continuation to run asynchronously is if there is a thread abort pending on the thread completing the antecedent task. It could be dangerous for TPL to continue running arbitrary amounts of work on a thread that the CLR is trying to tear down, so when a task completes and a thread abort is pending, all ContinueWith continuations will be queued rather than executed. Here’s a code sample to demonstrate this behavior:

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

class Program
{
    static void Main()
    {
        var tcs = new TaskCompletionSource<bool>();
        var cont = tcs.Task.ContinueWith(delegate
        {
            Console.WriteLine(Thread.CurrentThread.ManagedThreadId);
        }, TaskContinuationOptions.ExecuteSynchronously);

        new Thread(() =>
        {
            Console.WriteLine(Thread.CurrentThread.ManagedThreadId);
            try
            {
                Thread.CurrentThread.Abort();
            }
            finally
            {
                tcs.SetResult(true);
            }
        }).Start();

        cont.Wait();
    }
}

If you try running this code, you’ll find that it outputs two different thread IDs, one for the thread that’s completing the antecedent task and one for the thread that’s running the continuation. If you then comment out the line that’s issuing the Abort, you’ll find that it’ll output the same ID twice, since the continuation is then running synchronously on the same thread that completed the antecedent task.

Another condition that may force a continuation to run asynchronously has to do with stack overflows. TPL has logic used in a few places (e.g. Wait inlining, ExecuteSynchronously) to determine whether it’s too deep on the stack to safely execute. If TPL detects that it’s too close to the thread’s guard page for comfort, such that running the continuation synchronously has a high risk of overflowing, then TPL will force the continuation to run asynchronously. We can see this behavior with the following code example:

using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;

class Program
{
    static void Main()
    {
        var tcs = new TaskCompletionSource<bool>();
        var t = (Task)tcs.Task;

        var counts = new List<int>();
        int lastThreadId = Thread.CurrentThread.ManagedThreadId;
        int curCount = 0;

        for (int i = 0; i < 1000000; i++)
        {
            t = t.ContinueWith(delegate
            {
                int threadId = Thread.CurrentThread.ManagedThreadId;
                if (threadId == lastThreadId)
                {
                    curCount++;
                }
                else
                {
                    lastThreadId = threadId;
                    counts.Add(curCount);
                    curCount = 0;
                }
            }, TaskContinuationOptions.ExecuteSynchronously);
        }
        tcs.SetResult(true);
        t.Wait();
        Console.WriteLine((int)counts.Average());
    }
}

Here, we’re creating a chain of a million ExecuteSynchronously continuations. With a default stack size of 1MB, and with multiple stack frames require per continuation execution, this program would certainly overflow the stack if TPL didn’t account for it. However, if we run this, we should see that it successfully executes without crashing. Note that this code is keeping track of how many continuations contiguously run on the same thread so that we can see approximately how many continuations end up running synchronously before one is forced to run asynchronously in order to avoid an overflow (this isn’t a 100% accurate measure, as it’s theoretically possible the continuation could be queued to run asynchronously but then end up running on the same thread, if the thread unwound the whole stack and then picked up the continuation task to run it before any other thread did, but for our purposes it’s good enough). When I run this, I see that on average we run approximately 2200 continuations synchronously before we force one to run asynchronously.

The third current condition where ExecuteSynchronously continuations won’t run synchronously is when the target scheduler doesn’t allow it. A TaskScheduler has the ability to say whether tasks are able to run on the current thread or not. For example, if you created a custom StaTaskScheduler whose job was to run tasks on STA threads, it shouldn’t allow a task to run on an MTA thread, so that scheduler should be able to override the ExecuteSynchronously behavior in the case where the antecedent isn’t completing on an STA thread. When TPL goes to run a synchronous continuation, it does so via the target TaskScheduler’s TryExecuteTaskInline method. The scheduler can either choose to execute the task in that call, or it can choose to return false, meaning that it refused to run the task, and in the latter case, TPL will then queue the continuation to the scheduler to run asynchronously. Here’s an example:

using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;

class Program
{
    static void Main()
    {
        Console.WriteLine(Thread.CurrentThread.ManagedThreadId);
        var tcs = new TaskCompletionSource<bool>();
        var cont = tcs.Task.ContinueWith(delegate
        {
            Console.WriteLine(Thread.CurrentThread.ManagedThreadId);
        }, CancellationToken.None, TaskContinuationOptions.ExecuteSynchronously, new DummyTaskScheduler());
        tcs.SetResult(true);
        cont.Wait();
    }
}

class DummyTaskScheduler : TaskScheduler
{
    protected override void QueueTask(Task task)
    {
        ThreadPool.QueueUserWorkItem(delegate { TryExecuteTask(task); });
    }

    protected override bool TryExecuteTaskInline(Task task, bool taskWasPreviouslyQueued)
    {
        //return TryExecuteTask(task);
        return false;
    }

    protected override IEnumerable<Task> GetScheduledTasks()
    {
        return null;
    }
}

Note that our dummy TaskScheduler’s TryExecuteTaskInline is returning false and not executing the task; thus, even though the continuation has ExecuteSynchronously specified, the continuation will still run asynchronously. As with the previous examples, you can see this by running the code, which will output two different thread IDs. If you then modify TryExecuteTaskInline by commenting out the existing “return false;” and uncommenting in the “return TryExecuteTask(task);” line, then the program should output the same thread ID twice.

FromAsync(asyncResult, …) vs FromAsync(beginMethod, …)

Stephen Toub - MSFT — Mon, 06 Feb 2012 21:25:52 GMT

The Task Parallel Library (TPL) provides a set of “FromAsync” helper methods that create a Task or a Task<TResult> to represent an invocation of an APM method pair, i.e. BeginXx / EndXx. There are, however, two different flavors among these overloads: ones that accept an IAsyncResult “asyncResult” as the first parameter, and ones that accept a Func<…,AsyncCallback,object,IAsyncResult> “beginMethod” as the first parameter. What’s the difference, and why would you choose one over the other?

The fundamental difference between the two is in how the endMethod delegate is invoked. When FromAsync is given a beginMethod delegate, it passes to that beginMethod an AsyncCallback that will invoke the endMethod. As with any APM implemention, it’s up to the BeginXx’s implementation to eventually invoke this AsyncCallback when the async operation has completed. In contrast, the IAsyncResult interface doesn’t provide any mechanism via which to provide a callback that will be invoked when the operation represented by the IAsyncResult is completed, so when FromAsync is handed an IAsyncResult, it needs to monitor the IAsyncResult’s WaitHandle.

Monitoring the IAsyncResult’s WaitHandle (as returned from its AsyncWaitHandle property) has a few implications. First, since the APM pattern is all about asynchrony, most IAsyncResult implementations strive to avoid allocating the WaitHandle, only doing so lazily if it’s actually requested. Thus, by accessing the IAsyncResult’s AsyncWaitHandle property, FromAsync is likely forcing allocation of a WaitHandle (and the associated operating system primitives) that otherwise wouldn’t need to be allocated. Second, the only way to “monitor” a WaitHandle to know when it’s been set is to wait on it, which is a potentially blocking operation. Obviously, FromAsync isn’t going to block the current thread by waiting on this handle (that wouldn’t be very async of it), so instead it relies on the ThreadPool. It could queue a work item to block on the handle, but that would end up burning a ThreadPool thread for each such call. Instead, FromAsync relies on ThreadPool.RegisterWaitForSingleObject, a function that is specifically geared towards using the ThreadPool to more efficiently wait for handles. According to the MSDN documentation, it does so by waiting on multiple handles from a single thread via WaitForMultipleObjects; however, there are a limit to how many handles it can wait on from each thread, so while this will end up burning fewer threads than if we blocked one thread per handle, it’s still far from ideal.

As such, if you have a choice between using FromAsync(asyncResult, …) and using FromAsync(beginMethod, …), pick the latter that uses a delegate instead of an IAsyncResult. When might you choose to use the former? The IAsyncResult overloads can still be useful in a few cases. For example, there are some BeginXx methods with signatures that don’t exactly conform to the beginMethod functions accepted by the built-in FromAsync overloads, e.g. they have more parameters than the built-in versions can handle. In such a case, you could choose to either write your own FromAsync method to accommodate the custom signature, or you could use an IAsyncResult overload, since it doesn’t care about the shape of the BeginXx method, only about its result.

Building a custom GetOrAdd method for ConcurrentDictionary

Stephen Toub - MSFT — Sun, 05 Feb 2012 04:14:37 GMT

I was recently asked by a developer about getting some additional information out of ConcurrentDictionary<TKey,TValue>’s GetOrAdd method.

As a reminder, GetOrAdd either returns the value for a key currently in the dictionary, or if that key doesn’t have a value, it adds a value for the key as dictated by either a TValue provided by the caller or by executing a Func<TKey,TValue> provided by the caller. It then returns the new value. However, it doesn’t tell the caller which happened; all the caller knows is that it’s handed back the value (existing or new) that was associated with the key. The developer wanted to know which occurred, as he needed to notify some other code that a new value had been added.

While the built-in GetOrAdd method doesn’t provide this, we can build our own GetOrAdd overload to provide this behavior, building it on top of the existing TryGetValue and TryAdd methods:

public static TValue GetOrAdd<TKey, TValue>(
    this ConcurrentDictionary<TKey, TValue> dict,
    TKey key, Func<TKey, TValue> generator,
    out bool added)
{
    TValue value;
    while (true)
    {
        if (dict.TryGetValue(key, out value))
        {
            added = false;
            return value;
        }

        value = generator(key);
        if (dict.TryAdd(key, value))
        {
            added = true;
            return value;
        }
    }
}

The approach here is straightforward. We try to get the key’s value in the dictionary. If we can, then we use an out parameter to indicate that a new value was not added and we return the value we retrieved. If we can’t, then we try to add a new value as generated by the provided Func<TKey,TValue>; assuming that succeeds, we note the addition via the out parameter and we return the newly added value. If the addition fails (which should only happen if another thread concurrently added an item since we called TryGetValue), then we loop around and try the whole process again.

Using the TryGetValue, TryAdd, and TryUpdate on ConcurrentDictionary<TKey,TValue>, it’s possible to build many variations of this kind of behavior.

Know Thine Implicit Allocations

Stephen Toub - MSFT — Sat, 04 Feb 2012 01:26:07 GMT

For .NET 4.5, we’ve invested quite a bit of effort into performance, and in particular for the Task Parallel Library (Joe Hoag wrote a good paper covering some of these improvements). We focused such effort on TPL because it is a core component used in async programming and at a foundational level for many libraries and applications. Small tweaks to the performance characteristics of TPL can thusly have a noticeable impact on the performance of these higher-level systems.

In particular, one of the areas we focused a lot of attention on is memory utilization: how many objects are we allocating, and how big are those objects. This is primarily due to the subsequent costs of garbage collection necessary to clean up those objects… the less we allocate, the less often the GC will need to run and the shorter it will take to run. Sometimes these allocations are clearly visible, in that you see a “new” expression in the code. But often these allocations are hidden.

The Task Parallel Library is written in C#, and C# has some incredibly useful features that help to make programming very productive. The implementation of those features, however, can sometimes result in allocations of which it’s easy to be blissfully unaware. The developers that wrote the C# compiler have done some great work to make the code generated by the compiler efficient, and they’ve implemented a variety of optimizations to help minimize allocations. Still, if you care about eking out the best performance possible and minimizing all possible allocations in the process, it can be very useful to understand the intricacies of the code the compiler generates such that you can bend the behavior better to your will.

In this post, I’ll highlight some of these behaviors we came across during our performance push for TPL, and I’ll demonstrate how you can work around them to help get the best performance possible.

WARNING: Everything I’m discussing in this post is an implementation detail based on one version of the C# compiler (that in .NET 4.5 Developer Preview). It’s certainly possible that these characteristics may change in a future release, as the C# compiler gets more and more optimizations added to it.

Closure and Delegate Allocations

Lambdas and anonymous methods are very powerful features that save the developer from having to write a lot of boilerplate code. Consider the Framework’s ThreadPool.QueueUserWorkItem method:

public static bool ThreadPool(WaitCallback callback);

With lambdas/anonymous methods, I can write code like the following (I’m not recommending this as an approach towards writing to a stream asynchronously, rather just using it as an example of passing state around):

public static void WriteOnPool(Stream stream, byte [] data)
{
    ThreadPool.QueueUserWorkItem(delegate
    {
        stream.Write(data, 0, data.Length);
    });
}

Before lambdas/anonymous methods were introduced, C# developers wanting to call a method that accepted a delegate would typically need to define a class to store their state, add a method to that class to do their processing, and then create a delegate that pointed to that method on that state instance, e.g.

public static void WriteOnPool(Stream stream, byte [] data)
{
    var state = new WriteOnPoolState();
    state.stream = stream;
    state.data = data;
    ThreadPool.QueueUserWorkItem(new WaitCallback(state.Invoke));
}

private sealed class WriteOnPoolState
{
    public Stream stream;
    public byte [] data;

    public void Invoke(object ignored)
    {
        stream.Write(data, 0, data.Length);
    }
}

Now in C#, when you use an anonymous method or lambda, the compiler actually ends up generating code almost identical to this on your behalf so that you no longer have to do so manually. Here’s a decompiled version of what gets generated for my previous example that used an anonymous method:

public static void WriteOnPool(Stream stream, byte[] data)
{
    var locals2 = new DisplayClass1();
    locals2.stream = stream;
    locals2.data = data;
    ThreadPool.QueueUserWorkItem(
        new WaitCallback(locals2.<WriteOnPool>b__0));
}

[CompilerGenerated]
private sealed class DisplayClass1
{
    public Stream stream;
    public byte[] data;

    public void <WriteOnPool>b__0(object param0)
    {
        this.stream.Write(this.data, 0, this.data.Length);
    }
}

The important thing to note here is that there are two allocations involved: one for the closure object (in my hand-written version, WriteOnPoolState, and in the compiler-generated version, DisplayClass1), and one for the delegate to the method on the closure. For a method that simply accepts a delegate, there’s really no way around two allocations: the delegate object needs to be allocated, and it needs to target an object containing the state (assuming that state needs to change for each invocation, this state object also needs to be allocated for each call).

Luckily, many library methods that accept delegates recognize the need to pass state into the method, and thus accept not only a delegate object but also a state object that will be passed into the delegate when it’s invoked. In fact, ThreadPool has an additional overload of QueueUserWorkItem that accepts both, and the overload we’ve been using above just passes in null; that’s why the WriteOnPoolState.Invoke and DisplayClass1.<WriteOnPool>b__0 methods both accept an object parameter, so as to match the required signature of a WaitCallback, which accepts an object parameter.

This pattern of accepting an object state is useful because it allows us to avoid allocations. The reason we need to allocate a delegate each time is because its Target needs to point to the right “this” object. But if there is no “this”, i.e. we’re dealing with a static method, then we can reuse the same delegate object over and over, simply passing in different state as a parameter each time. To achieve that, we can rewrite our WriteOnPool example as follows:

public static void WriteOnPool(Stream stream, byte [] data)
{
    var state = new WriteOnPoolState();
    state.stream = stream;
    state.data = data;
    ThreadPool.QueueUserWorkItem(s_writeOnPoolBody, state);
}

private static readonly WaitCallback s_writeOnPoolBody =
    new WaitCallback(WriteOnPoolBody);

private static void WriteOnPoolBody(object state)
{
    var poolState = (WriteOnPoolState)state;
    poolState.stream.Write(poolState.data, 0, poolState.data.Length);
}

private sealed class WriteOnPoolState
{
    public Stream stream;
    public byte [] data;
}

We’re now still allocating a delegate and a state object, but we allocate the delegate only once and cache it into a static field. On each call to WriteOnPool, then, we only allocate once, for the WriteOnPoolState object, saving half of the allocations.

As it turns out, the C# compiler has smarts to provide similar savings when using lambdas and anonymous methods. Let’s say I instead rewrite my implementation as follows, still using a lambda, but now passing in my state via the object state parameter and ensuring that my lambda doesn’t close over anything (I’m also using a Tuple<,> for conciseness to avoid needing to define my own custom state type):

public static void WriteOnPool(Stream stream, byte[] data)
{
    ThreadPool.QueueUserWorkItem(state =>
    {
        var poolState = (Tuple<Stream,byte[]>)state;
        poolState.Item1.Write(
            poolState.Item2, 0, poolState.Item2.Length);
    }, Tuple.Create(stream, data));
}

This now results in the following code from the compiler:

public static void WriteOnPool(Stream stream, byte[] data)
{
    if (CachedAnonymousMethodDelegate1 == null)
        CachedAnonymousMethodDelegate1 =
            new WaitCallback(Program.<WriteOnPool>b__0);

    ThreadPool.QueueUserWorkItem(
        CachedAnonymousMethodDelegate1,
        Tuple.Create<Stream, byte[]>(stream, data));
}

[CompilerGenerated]
private static WaitCallback CachedAnonymousMethodDelegate1;

[CompilerGenerated]
private static void <WriteOnPool>b__0(object state)
{
    Tuple<Stream, byte[]> poolState = (Tuple<Stream, byte[]>) state;
    poolState.Item1.Write(poolState.Item2, 0, poolState.Item2.Length);
}

Note that the compiler has done something very similar to what we did manually, creating a static field to cache the delegate. The primary difference between what we manually wrote and what the compiler generated is the compiler is lazily instantiating the delegate on first use, rather than doing it in the static constructor as we did by using a field initializer. So, we get to have our cake and eat most of it, too… we still get most of the convenience of lambdas while getting the efficiency benefits of the delegate caching. And, of course, if the state I’m trying to pass in is already a single object, that object can be passed in directly (rather than wrapping it in a state object), bringing us down to zero additional allocations.

When Automatic Caching Fails

This is all well and good, but it can lead us into a fall sense of security, as there are times when the compiler will not cache the delegate.

One prime example is that the compiler won’t cache the delegate when a method group is used rather than a lambda / anonymous method, e.g.

public void WriteOnPool(Stream stream, byte[] data)
{
    ThreadPool.QueueUserWorkItem(
        WriteOnPoolCallback,
        Tuple.Create(stream, data));
}

private static void WriteOnPoolCallback(object state)
{
    var poolState = (Tuple<Stream, byte[]>)state;
    poolState.Item1.Write(
        poolState.Item2, 0, poolState.Item2.Length);
}

Some developers prefer the explicitness of having a named method, but the compiler currently generates code like the following for this:

public void WriteOnPool(Stream stream, byte[] data)
{
    ThreadPool.QueueUserWorkItem(
        new WaitCallback(Program.WriteOnPoolCallback),
        Tuple.Create<Stream, byte[]>(stream, data));
}

private static void WriteOnPoolCallback(object state)
{
    Tuple<Stream, byte[]> poolState = (Tuple<Stream, byte[]>) state;
    poolState.Item1.Write(
        poolState.Item2, 0, poolState.Item2.Length);
}

Notice the delegate allocation on each call. In such a situation, you might prefer to do the caching manually, or to revert to using a lambda or anonymous method.

Another example where the compiler won’t cache is when dealing with certain patterns of generics usage. Consider the following variation. Here rather than writing a byte[] to the stream, I’m using a BinaryFormatter to serialize an arbitrary generic T to the stream:

public void WriteOnPool<T>(Stream stream, T data)
{
    ThreadPool.QueueUserWorkItem(state =>
    {
        var poolState = (Tuple<Stream, T>)state;
        new BinaryFormatter().Serialize(
            poolState.Item1, poolState.Item2);
    }, Tuple.Create(stream, data));
}

The compiler will currently not cache a delegate for this case, instead generating code like the following:

public void WriteOnPool<T>(Stream stream, T data)
{
    ThreadPool.QueueUserWorkItem(
        new WaitCallback(Program.<WriteOnPool>b__0<T>),
        Tuple.Create<Stream, T>(stream, data));
}

[CompilerGenerated]
private static void <WriteOnPool>b__0<T>(object state)
{
    Tuple<Stream, T> poolState = (Tuple<Stream, T>) state;
    new BinaryFormatter().Serialize(
        poolState.Item1, poolState.Item2);
}

A little thought reveals why the compiler doesn’t use the same caching pattern it does elsewhere. Notice that the method generated for the lambda is a generic method, and for it to be cached, it would need to be cached in a static field for each T that might exist… that’s not possible in a field on a class that’s not parameterized on the same generic type parameter. We can manually work around this by caching the delegate manually: we just need to create a generic type to contain the field:

public void WriteOnPool<T>(Stream stream, T data)
{
    ThreadPool.QueueUserWorkItem(
        DelegateCache<T>.s_writeOnPoolCallback,
        Tuple.Create(stream, data));
}

private static class DelegateCache<T>
{
    internal static readonly WaitCallback s_writeOnPoolCallback =
        state =>
    {
        var poolState = (Tuple<Stream, T>)state;
        new BinaryFormatter().Serialize(
            poolState.Item1, poolState.Item2);
    };
}

Theoretically, the compiler could perform a similar transformation, for example creating one such generic cache per assembly and per generic arity, but it currently does not do so.

Another common case where such lack-of-caching goes unnoticed is when dealing with instance methods and data. Consider the following similar code:

internal class MyCoolType
{
    private Stream m_stream;
    private byte[] m_data;

    public void WriteOnPool()
    {
        ThreadPool.QueueUserWorkItem(delegate
        {
            m_stream.Write(m_data, 0, m_data.Length);
        });
    }
}

This example will generate code like the following:

internal class MyCoolType
{
    private Stream m_stream;
    private byte[] m_data;

    public void WriteOnPool()
    {
        ThreadPool.QueueUserWorkItem(
            new WaitCallback(this.<WriteOnPool>b__0));
    }

    [CompilerGenerated]
    private void <WriteOnPool>b__0(object param0)
    {
        this.m_stream.Write(this.m_data, 0, this.m_data.Length);
    }
}

The anonymous method refers implicitly to ‘this’, closing over member fields of the current instance. As such, the compiler generates an instance method for that anonymous method, and then because it needs a delegate instance with a Target set to this particular instance, each call to WriteOnPool will end up allocating a new delegate (it would be possible for the compiler to have added an instance field in order to cache this delegate, but that could have negative consequences, such as making each instance of this class bigger, and the compiler can’t easily reason about the global impacts of such caching). Of course, we have knowledge that the compiler doesn’t have: we know that we can use the object state parameter in order to pass data into the method. So, we could instead write our code as:

internal class MyCoolType
{
    private Stream m_stream;
    private byte[] m_data;

    public void WriteOnPool()
    {
        ThreadPool.QueueUserWorkItem(state =>
        {
            MyCoolType mct = (MyCoolType)state;
            mct.m_stream.Write(mct.m_data, 0, mct.m_data.Length);
        }, this);
    }
}

or as:

internal class MyCoolType
{
    private Stream m_stream;
    private byte[] m_data;

    public void WriteOnPool()
    {
        ThreadPool.QueueUserWorkItem(
            state => ((MyCoolType)state).Write(), this);
    }

    private void Write()
    {
        m_stream.Write(m_data, 0, m_data.Length);
    }
}

and in both cases, we’ll get the caching we expect. This is an example of the case I referred to earlier, where we already have a single state object to pass in, and thus we don’t need to allocate a wrapper (e.g. a tuple), which means there are no additional allocations here.

Closures and Fast Paths

There are other times when the compiler isn’t yet quite as clever as we might expect it to be. One case in which this comes up is when using “fast paths” to optimize frequented code paths. Consider a change to our WriteOnPool method: the goal of the method isn’t to queue a work item to do the write, but rather to ensure the write happens on a ThreadPool thread; as such, if we’re already on a ThreadPool thread, we can avoid the queueing operation:

public static void WriteOnPool(Stream stream, byte[] data)
{
    if (Thread.CurrentThread.IsThreadPoolThread)
    {
        stream.Write(data, 0, data.Length);
    }
    else
    {
        ThreadPool.QueueUserWorkItem(delegate
        {
            stream.Write(data, 0, data.Length);
        });
    }
}

I might then expect that my “fast path” that occurs when I’m already on a ThreadPool thread won’t involve any additional allocations for the closure/delegate used on the “slow path.” But such an expectation is based on an assumption that the compiler allocates the closure in the “else” block. In fact, the compiler currently generates code like the following:

public static void WriteOnPool(Stream stream, byte[] data)
{
    var locals3 = new DisplayClass2();
    locals3.stream = stream;
    locals3.data = data;
    if (Thread.CurrentThread.IsThreadPoolThread)
    {
        locals3.stream.Write(locals3.data, 0, locals3.data.Length);
    }
    else
    {
        ThreadPool.QueueUserWorkItem(
            new WaitCallback(locals3.<WriteOnPool>b__0));
    }
}

[CompilerGenerated]
private sealed class DisplayClass2
{
    public byte[] data;
    public Stream stream;

    public void <WriteOnPool>b__0(object param0)
    {
        this.stream.Write(this.data, 0, this.data.Length);
    }
}

Note that the “locals” that are captured into the closure actually live on the heap-allocated closure object and are accessed there, even from code paths that don’t use the closure. Thus, the closure object gets allocated at the beginning of the method. This means that we’re still paying for the closure object even if we only take the “fast path” that doesn’t actually require it. If we instead separate out the closure into a separate method, we can eliminate that allocation from the fast path, e.g.

public static void WriteOnPool(Stream stream, byte[] data)
{
    if (Thread.CurrentThread.IsThreadPoolThread)
    {
        stream.Write(data, 0, data.Length);
    }
    else WriteOnPoolPrivate(stream, data);
}

private static void WriteOnPoolPrivate(Stream stream, byte[] data)
{
    ThreadPool.QueueUserWorkItem(delegate
    {
        stream.Write(data, 0, data.Length);
    });
}

Moral of the Story

C# and Visual Basic provide some very powerful language features for you to use, and they typically do a very good job of implementing what’s under the hood. But if your particular scenario demands pedal-to-the-medal performance, you need to at least understand what’s happening on your behalf so that you can better influence it as necessary. Get friendly with tools like .NET Reflector that help you to see what’s actually going on in the compiled code.

If you do use a decompiler for this kind of analysis, be certain that it’s not sweeping such details under the rug for you. A primary use of these decompilers is to get high-quality source generated from low-level IL, and thus decompilers will often attempt to decompile back to source that employs anonymous methods and lambdas and the like. If instead your goal is to get a closer view on what’s actually happening, try tweaking the settings of the decompiler to minimize what language features it reverse engineers to, e.g. in .NET Reflector, you can change the “optimization level” so that it shows you code like what I’ve shown in this post, rather than hiding these allocations behind a lambda.

One last thought. Throughout this post, I kept harping on the number of allocations, but I didn’t spend much time on the size of the allocations. The number of allocations is only one aspect to consider. We also want to consider the amount of memory being allocated, as that will factor into how often the garbage collector runs. There are lots of tools to help determine the amount of memory .NET applications are allocating, but when prototyping I often find myself using a function like the following to give me a rough sense for what something is costing:

public static class MemoryAnalysis
{
    public static long AveragedApproximateSize<T>(Func<T> rootGenerator) where T : class
    {
        const int iters = 1000000;

        var items = new object[iters];
        long start = GC.GetTotalMemory(true);

        for (int i = 0; i < items.Length; i++)
            items[i] = rootGenerator();

        long end = GC.GetTotalMemory(true);
        GC.KeepAlive(items);

        return (long)Math.Round((end - start) / (double)iters);
    }
}

When I then run:

static void Main()
{
    Console.WriteLine(MemoryAnalysis.AveragedApproximateSize(
        () => new WriteOnPoolState()));
    Console.WriteLine(MemoryAnalysis.AveragedApproximateSize(
        () => new WaitCallback(WriteOnPoolBody)));
}

I see that WriteOnPoolState is 16 bytes and WaitCallback is 32 bytes (in a 32-bit compilation), which makes logical sense if you add up the size of the fields on the objects plus the object headers. That means that for the examples where we were able to eliminate the delegate allocation by caching the delegate in a static, we were not only saving half the number of allocations, but we were also saving 2/3^rds of the memory being allocated.

Happy optimizing.

Await, SynchronizationContext, and Console Apps: Part 3

Stephen Toub - MSFT — Fri, 03 Feb 2012 06:46:00 GMT

In Part 1 and Part 2 of this short series, I demonstrated how you can build a SynchronizationContext and use it run an async method such that all of the continuations in that method will run on serialized on the current thread. This can be helpful when executing async methods in a console app, or in a unit test framework that doesn’t directly support async methods. However, the support I showed thus far targets async methods that return Task… what about async methods that return void?

C# and Visual Basic support two flavors of async methods: ones that return tasks (either Task or Task<T>) and ones that return void. The former use the returned Task to represent the completion of the async method. In the case of an “async void” method, however, there is no returned Task to represent the method’s processing. Instead, “async void” methods interact with the current SynchronizationContext to alert the context to the async method’s execution status. Before entering the body of the async method, if there is a current SynchronizationContext, it is retrieved and its OperationStarted method is called. And after the async method has completed, that same context has its OperationCompleted method called. Further, if an exception goes unhandled in the body of the async void method, the throwing of that exception is Post to the SynchronizationContext, so that the exception escapes back to the context for it to handle as it pleases.

All of this means that if we want our AsyncPump to be able to handle “async void” methods in addition to “async Task” methods, we need to augment the type slightly. First, we need to augment our SingleThreadSynchronizationContext to react appropriate to calls to OperationStarted and OperationCompleted. These methods need to maintain a count of how many outstanding operations there are, such that when the count reaches 0, we call Complete, just as before we called Complete when the async method’s Task completed. We do this by adding three members to the custom context:

private int m_operationCount = 0;

public override void OperationStarted()
{
    Interlocked.Increment(ref m_operationCount);
}

public override void OperationCompleted()
{
    if (Interlocked.Decrement(ref m_operationCount) == 0)
        Complete();
}

Then we need to add a new AsyncPump.Run overload that works with Action (for “async void” methods) instead of with Func<Task> (for “async Task” methods). As a reminder, here’s the existing Run method from our AsyncPump class:

public static void Run(Func<Task> asyncMethod)
{
    var prevCtx = SynchronizationContext.Current;
    try
    {
        var syncCtx = new SingleThreadSynchronizationContext();
        SynchronizationContext.SetSynchronizationContext(syncCtx);

        var t = asyncMethod();
        t.ContinueWith(delegate { syncCtx.Complete(); }, TaskScheduler.Default);

        syncCtx.RunOnCurrentThread();
        t.GetAwaiter().GetResult();
    }
    finally
    {
        SynchronizationContext.SetSynchronizationContext(prevCtx);
    }
}

Most of this will remain the same for our new Action-based variant. In fact, for the new one, we primarily just need to delete all the code having to do with the returned task, since there isn’t one, and all notion of completion is being handled by the OperationStarted and OperationCompleted methods we added to the context. We do surround the asyncMethod invocation with calls to OperationStarted and OperationCompleted, just in case the asyncMethod is actually just a void method and not an “async void” method, in which case we need to make sure the operation count is greater than 0 for the duration of the invocation in order to avoid races that could result if the delegate invoked other async void methods.

public static void Run(Action asyncMethod)
{
    var prevCtx = SynchronizationContext.Current;
    try
    {
        var syncCtx = new SingleThreadSynchronizationContext();
        SynchronizationContext.SetSynchronizationContext(syncCtx);

        syncCtx.OperationStarted();
        asyncMethod();
        syncCtx.OperationCompleted();

        syncCtx.RunOnCurrentThread();
    }
    finally
    {
        SynchronizationContext.SetSynchronizationContext(prevCtx);
    }
}

That’s it. We’re now able to use our AsyncPump to run “async void” methods synchronously with all continuations executed on the current thread, e.g.

static void Main()
{
    AsyncPump.Run((Action)FooAsync);
}

static async void FooAsync()
{
    Foo1Async();
    await Foo2Async();
    Foo3Async();
}

static async void Foo1Async()
{
    await Task.Delay(1000);
    Console.WriteLine(1);
}

static async Task Foo2Async()
{
    await Task.Delay(1000);
    Console.WriteLine(2);
}

static async void Foo3Async()
{
    await Task.Delay(1000);
    Console.WriteLine(3);
}

Happy async’ing.

Advanced APM Consumption in Async Methods

Stephen Toub - MSFT — Mon, 23 Jan 2012 23:39:24 GMT

I’ve previously blogged about how to expose existing Asynchronous Programming Model (APM) implementations as Task-based methods. This can be done manually using a TaskCompletionSource<TResult>, or it can be done using the built-in wrapper provided in TPL via the Task.Factory.FromAsync method. By creating a Task-based wrapper for such BeginXx/EndXx method pairs, APM implementations become consumable in async methods in C# and Visual Basic: since Tasks are awaitable, any operation exposed as a Task is then inherently awaitable.

What I haven’t before discussed is how you can consume an APM implementation without wrapping it in a Task, and that’s what I’ll do in this blog post. What follows is an advanced approach with the only benefit being potentially reduced overheads, and the resulting performance gains are really only relevant in extreme situations. But such situations do exist, primarily in high-throughput server scenarios. Further, the techniques involved here are interesting regardless of whether you decide to use them in practice (I previously discussed a similar approach specifically geared towards sockets). Without further ado…

The key to this approach is that the C# and Visual Basic compilers support awaiting things other than Tasks, and in fact support awaiting anything that follows the right pattern. We can take advantage of that to implement a custom awaitable geared explicitly towards optimizing performance when consuming APM implementations. For very high-throughput workloads, allocations and the resulting garbage collections can lead to noticeable overheads, and thus we want to minimize allocations as much as possible. In particular, there are many APM implementations that can complete their processing synchronously (i.e. during the BeginXx method), e.g. if you asynchronously retrieve a value that happens to already be buffered or cached, it’s possible to synchronously satisfy the request from that buffer or cache, rather than performing I/O to retrieve or generate it. For situations where it’s possible for requests to be satisfied synchronously, it’s common that these synchronous code paths become hot paths, and as such we want to avoid as much overhead as possible on them.

One of the tricks I’ve mentioned in the past for minimizing allocations is caching and reusing the objects used to represent asynchronous operations. For example, in the case of async methods with C# and Visual Basic, the async method infrastructure caches common Task and Task<TResult> instances that it can use repeatedly (if the right conditions are met) as the return object from an async method. As another example (and an internal implementation detail, so this could of course change in the future), MemoryStream’s ReadAsync method will remember the last Task<int> it returned for a successful read, and if the subsequent ReadAsync operation completes with the same number of bytes, it can reuse and return that same cached Task<int> rather than instantiating a new one. We can take advantage of this same kind of optimization in order to minimize allocations when consuming APM implementations directly.

To do so, we’ll develop an “AsyncApmAdapter” type that implements the await pattern and that we can reuse over and over again for sequential APM calls. This type will be awaitable, with its GetResult method returning the IAsyncResult for the operation. We’ll pass an adapter instance as the object state into the BeginXx method, along with an AsyncCallback that will configure the adapter instance appropriately when the asynchronous operation completes. We’ll then await the adapter and pass the resulting IAsyncResult into the EndXx method. This yields the following pattern:

var adapter = new AsyncApmAdapter();
...
BeginXx(..., AsyncApmAdapter.Callback, adapter);
EndXx(await adapter);

To use my favorite looping example, if we wanted to write a read / write loop to copy from one stream to another using Stream.Begin/EndRead and Stream.Begin/EndWrite, that could look like:

static async Task CopyStreamAsync(Stream source, Stream dest)
{
    var adapter = new AsyncApmAdapter();
    var buffer = new byte[0x1000];

    while(true)
    {
        source.BeginRead(buffer, 0, buffer.Length,
            AsyncApmAdapter.Callback, adapter);
        int numRead = source.EndRead(await adapter);
        if (numRead == 0) break;

        dest.BeginWrite(buffer, 0, numRead,
            AsyncApmAdapter.Callback, adapter);
        dest.EndWrite(await adapter);
    }
}

Note that we’ve only had to allocate the one AsyncApmAdapter instance, which we can reuse over and over again as we make Begin/EndRead and Begin/EndWrite APM calls to the source and destination streams.

How do we achieve such magic? Let’s start by defining our adapter type, which will act both as an awaitable and as an awaiter:

internal sealed class AsyncApmAdapter
{
    public AsyncApmAdapter GetAwaiter() { return this; }

    public bool IsCompleted { get { ... } }
    public void OnCompleted(Action continuation) { ... }
    public IAsyncResult GetResult() { ... }

    ... // fields
}

We’ll need a few fields in our adapter. First, we’ll need to be able to store an IAsyncResult for the APM side of things, and second, we’ll need to be able to store the Action continuation delegate that will be supplied by the async method when awaiting on our adapter instance. We’re also going to need a dummy Action delegate that we can use as a sentinel, a marker to help synchronize (we’ll see where and why shortly).

private IAsyncResult _asyncResult;
private Action _continuation;
private readonly static Action CALLBACK_RAN = () => { };

As used in the earlier example, our adapter exposes a static Callback field of type AsyncCallback; by creating and caching this delegate once, we can use it for all invocations (it doesn’t mutate any shared state). It is the responsibility of this AsyncCallback to mutate supplied the adapter instance (passed in via the object state) when the async operation completes. If the operation completes synchronously (such that CompletedSynchronously is true), the only thing the callback needs to do is store the IAsyncResult for the operation into the adapter. If, however, the operation completes asynchronously, then we need to do a bit more.

public readonly static AsyncCallback Callback = asyncResult =>
{
    var adapter = (AsyncApmAdapter) asyncResult.AsyncState;
    adapter._asyncResult = asyncResult;

    if (!asyncResult.CompletedSynchronously)
    {
        Action continuation = adapter._continuation ??
            Interlocked.CompareExchange(
                ref adapter._continuation, CALLBACK_RAN, null);
        if (continuation != null) continuation();
    }
};

In the case where we’re completing asynchronously, we’re potentially racing with the call site that will be checking whether the async method should continue running synchronously (via the awaiter’s IsCompleted) or whether it should yield, suspending the async method’s execution (via the awaiter’s OnCompleted). As such, there’s some coordination that needs to happen between the callback and IsCompleted / OnCompleted. The OnCompleted method will set the _continuation field to the Action provided by async method infrastructure. As such, if _continuation is already non-null, then OnCompleted was already invoked, and we thus need to invoke the supplied Action _continuation here. If, however, _continuation is null, we need to set it to our sentinel Action delegate (CALLBACK_RUN)… this will indicate to IsCompleted / OnCompleted that the async operation has already finished, and thus IsCompleted / OnCompleted needs to take care of continuing the execution.

IsCompleted’s job is relatively simple. The _asyncResult field starts out on the adapter as null. If it’s still null when IsCompleted is accessed, then we must yield so that execution can continue asynchronously. If the _asyncResult field is non-null when IsCompleted is accessed, that means the callback has at least started running (and filled in _asyncResult as part of that), but we don’t know whether it’s finished or not. So, we check one of two additional conditions. If the IAsyncResult’s CompletedSynchronously is true, then we know the callback completed, because it was invoked synchronously as part of the call to BeginXx, and we’re not accessing IsCompleted until after BeginXx returns. If, however, CompletedSynchronously is false, then we’ll check the _continuation field itself. Remember that the callback will try to set the _continuation field to the CALLBACK_RAN sentinel, so here in IsCompleted if _continuation is already CALLBACK_RAN, then we know that, well, the callback ran.

public bool IsCompleted
{
    get
    {
        IAsyncResult iar = _asyncResult;
        return iar != null &&
               (iar.CompletedSynchronously ||
                _continuation == CALLBACK_RAN);
    }
}

There are some benign races here. It’s possible that the async operation could be completing concurrently with IsCompleted being called, and that’s ok… in the rare case where the operation has already completed but where IsCompleted doesn’t notice, it’ll simply return false, and we’ll just rely on OnCompleted to handle synchronizing with the callback.

Now on to OnCompleted. In the await pattern, OnCompleted is invoked to schedule the Action continuation delegate. To do so in our case, ideally OnCompleted can just store the provided Action into the _continuation field, such that when the async operation completes, the callback will invoke _continuation (as we saw earlier). However, it’s possible that IsCompleted returns false, but concurrently with invoking IsCompleted and OnCompleted, the operation then completes. Thus, OnCompleted must be able to deal with the case where the operation has already completed by the time OnCompleted is invoked. However, OnCompleted must not invoke the delegate synchronously, even if it determines that the async operation has already completed, otherwise we risk “stack dives”. So, OnCompleted checks whether the callback has already tried to pick up the _continuation delegate, which OnCompleted knows by seeing if _continuation is CALLBACK_RAN. If it’s not yet CALLBACK_RAN, then it’ll simply store the supplied continuation delegate into _continuation. If, however, it is already CALLBACK_RAN, then it’ll use a Task to schedule the continuation to run asynchronously. And, for correctness, any writes here need to happen as part of an atomic compare-and-swap operation.

public void OnCompleted(Action continuation)
{
    if (_continuation == CALLBACK_RAN ||
        Interlocked.CompareExchange(
            ref _continuation, continuation, null) == CALLBACK_RAN)
    {
        Task.Run(continuation);
    }
}

Finally, we need to implement GetResult. As I mentioned previously, GetResult’s role will be to return the stored IAsyncResult. However, we also want to be able to reuse this adapter instance over and over, which means we need to reset its state after each operation completes, and GetResult is a good place to do that:

public IAsyncResult GetResult() {
    IAsyncResult result = _asyncResult;
    _asyncResult = null;
    _continuation = null;
    return result;
}

That’s it for the implementation. A bit tricky, certainly, but relatively little code, and it can provide for improved performance in those extreme situations I alluded to earlier. Of course, that potential overhead reduction comes at the price of readability. For example, with this approach you can no longer use the async operation as an expression. To address that, a colleague of mine, Eric Eilebrecht, suggested a cute hack: implement a GetAwaiter extension method on IAsyncResult:

public static class IAsyncResultExtensions
{
    public static AsyncApmAdapter GetAwaiter(this IAsyncResult iar)
    {
        return (AsyncApmAdapter)iar.AsyncState;
    }
}

With that, instead of writing:

BeginXx(..., AsyncApmAdapter.Callback, adapter);
EndXx(await adapter);

you could write:

EndXx(await BeginXx(..., AsyncApmAdapter.Callback, adapter));

thereby regaining some of the expressiveness lost in the approach. Of course, this extension method only makes sense when the object state is an AsyncApmAdapter used in this manner, so be careful with such an approach.

Per my comments at the beginning of this post, if you need to consume an APM implementation, converting it into a Task (via Task.Factory.FromAsync or other means) is a great way. However, if you do find yourself in a position where you’re using an async method to consume an APM implementation, and if it’s crucial to your scenario that you eek out every last bit of performance possible, you could consider an approach like that outlined in this post. As with any micro-optimization that leads to (slightly) less readable code, make sure it’s worthwhile before paying for it, and measure, measure, measure.

Enjoy!

Await, SynchronizationContext, and Console Apps: Part 2

Stephen Toub - MSFT — Sun, 22 Jan 2012 00:49:09 GMT

Yesterday, I blogged about how you can implement a custom SynchronizationContext in order to pump the continuations used by async methods so that they may be processed on a single, dedicated thread. I also highlighted that this is basically what UI frameworks like Windows Forms and Windows Presentation Foundation do with their message pumps.

Now that we understand the mechanics of how these things work, it’s worth pointing out that we can achieve the same basic semantics without writing our own custom SynchronizationContext. Instead, we can use one that already exists in the .NET Framework: DispatcherSynchronizationContext. Through its Dispatcher class and its PushFrame method, WPF provides the ability to (as described in MSDN) “enter an execute loop” that “processes pending work items.” This is exactly what our custom SynchronizationContext was doing with its usage of BlockingCollection<T>, so we can just use WPF’s support instead of developing our own. (You might ask then why I started by describing how to do it manually and writing my own to exemplify it. I do so because I think it’s important to really understand how things work; I typically find that developers write better higher-level code if they have the right mental model for what’s happening under the covers, allowing them to better reason about bugs, about performance, about reliability, and the like.)

Below you can see how few lines of code it takes to achieve this support. (To compile this code, you’ll need to reference WindowsBase.dll to bring in the relevant WPF types.)

using System;
using System.Threading;
using System.Threading.Tasks;
using System.Windows.Threading;

public static class AsyncPump
{
    public static void Run(Func<Task> func)
    {
        if (func == null) throw new ArgumentNullException("func");

        var prevCtx = SynchronizationContext.Current;
        try
        {
            var syncCtx = new DispatcherSynchronizationContext();
            SynchronizationContext.SetSynchronizationContext(syncCtx);

            var t = func();
            if (t == null) throw new InvalidOperationException();

            var frame = new DispatcherFrame();
            t.ContinueWith(_ => { frame.Continue = true; },
                TaskScheduler.Default);
            Dispatcher.PushFrame(frame);

            t.GetAwaiter().GetResult();
        }
        finally
        {
            SynchronizationContext.SetSynchronizationContext(prevCtx);
        }
    }
}

In short, we instantiate a DispatcherSynchronizationContext and publish it to be Current on the current thread; this is what enables the awaits inside of the async method being processed to queue their continuations back to this thread and this thread’s Dispatcher. Then we instantiate a DispatcherFrame, which represents an execute loop for WPF’s message pump. We use a continuation to signal to that DispatcherFrame when it should exit its loop (i.e. when the Task completes). And then we start the frame’s execute loop via the PushFrame method.

All in all, just a few lines of code to achieve some powerful and useful behavior.

Implementing a SynchronizationContext.SendAsync method

Stephen Toub - MSFT — Sat, 21 Jan 2012 00:23:00 GMT

I recently saw two unrelated questions, the answers to which combine to form a potentially useful code snippet.

The first question was about SynchronizationContext. SynchronizationContext provides a Post method, which asynchronously schedules the supplied delegate and object state to be executed according to the SynchronizationContext’s whims. However, Post returns void, so there’s no built-in way to know when the operation has completed. SynchronizationContext also provides a Send method, which won’t return until the delegate has been executed; that means that while you do have a built-in way to know when the operation has completed, it’s also a synchronous operation, blocking the current thread. The question then was whether there was a way to achieve the asynchronous behavior of Post but with the ability of Send to know when the operation has completed, e.g. an asynchronous Send:

public Task SendAsync(SendOrPostCallback d, object state);

The second question was about implementing asynchronous operations that return Tasks but without using async and await. In effect, the questioner wanted to create a Task-returning method to represent an existing non-Task-based mechanism they were already using.

So, let’s try to satisfy both questions by implementing SendAsync using Post as the underlying mechanism that we’ll wrap with a Task.

The first thing we need is a TaskCompletionSource<T>. TaskCompletionSource<T> is the mechanism within the Task Parallel Library that exists to create tasks that you can explicitly complete. Rather than creating a Task that’s associated with a delegate and whose completion is then tied to that delegate’s execution, TaskCompletionSource<T> create as task that’s not associated with a delegate, and it gives you methods like SetResult and SetException that allow you to complete the task when and how you see fit.

This capability makes TaskCompletionSource ideal for wrapping existing operations. The idea is that we create a TaskCompletionSource<T> and we then use its Task property to get at the task that we can manually complete via the TaskCompletionSource. We then launch the underlying operation being wrapped, and when it completes, we either invoke the SetResult method with the result of the operation completing successfully, or we invoke the SetException method with the error that resulted from its failure. We can follow this pattern to implement SendAsync by wrapping Post:

public static Task SendAsync(
    this SynchronizationContext context, SendOrPostCallback d, object state)
{
    var tcs = new TaskCompletionSource<bool>();
    context.Post(delegate {
        try {
            d(state);
            tcs.SetResult(true);
        }
        catch(Exception e) { tcs.SetException(e); }
    }, null);
    return tcs.Task;
}

Here, the delegate we pass to Post invokes the original delegate, but it does so wrapped in a try/catch block. If the invocation of the original delegate succeeds, we complete the Task successfully, and if an exception occurred from invoking the delegate, we complete the task with the thrown error. With this extension method on SynchronizationContext, we can now write code that uses await to asynchronously wait for an operation sent to a particular SynchronizationContext to complete. And while there are various things we could do to improve the performance of this method slightly (e.g. by taking advantage of Post’s object state parameter rather than using a closure), overall it’s a fine representation of how you can create Task-based wrappers for existing mechanisms.

Await, SynchronizationContext, and Console Apps

Stephen Toub - MSFT — Fri, 20 Jan 2012 23:00:00 GMT

When I discuss the new async language features of C# and Visual Basic, one of the attributes I ascribe to the await keyword is that it “tries to bring you back to where you were.” For example, if you use await on the UI thread of your WPF application, the code that comes after the await completes should run back on that same UI thread.

There are several mechanisms that are used by the async/await infrastructure under the covers to make this marshaling work: SynchronizationContext and TaskScheduler. While the transformation is much more complicated than what I’m about to show, logically you can think of the following code:

await FooAsync();

RestOfMethod();

as being similar in nature to this:

var t = FooAsync();

var currentContext = SynchronizationContext.Current;

t.ContinueWith(delegate

{

if (currentContext == null)

RestOfMethod();

else

currentContext.Post(delegate { RestOfMethod(); }, null);

}, TaskScheduler.Current);

In other words, before the async method yields to asynchronously wait for the Task ‘t’, we capture the current SynchronizationContext. When the Task being awaited completes, a continuation will run the remainder of the asynchronous method. If the captured SynchronizationContext was null, then RestOfMethod() will be executed in the original TaskScheduler (which is often TaskScheduler.Default, meaning the ThreadPool). If, however, the captured context wasn’t null, then the execution of RestOfMethod() will be posted to the captured context to run there.

Both SynchronizationContext and TaskScheduler are abstractions that represent a “scheduler”, something that you give some work to, and it determines when and where to run that work. There are many different forms of schedulers. For example, the ThreadPool is a scheduler: you call ThreadPool.QueueUserWorkItem to supply a delegate to run, that delegate gets queued, and one of the ThreadPool’s threads eventually picks up and runs that delegate. Your user interface also has a scheduler: the message pump. A dedicated thread sits in a loop, monitoring a queue of messages and processing each; that loop typically processes messages like mouse events or keyboard events or paint events, but in many frameworks you can also explicitly hand it work to do, e.g. the Control.BeginInvoke method in Windows Forms, or the Dispatcher.BeginInvoke method in WPF.

SynchronizationContext, then, is just an abstract class that can be used to represent such a scheduler. The base class exposes several virtual methods, but we’ll focus on just one: Post. Post accepts a delegate, and the implementation of Post gets to decide when and where to run that delegate. The default implementation of SynchronizationContext.Post just turns around and passes it off to the ThreadPool via QueueUserWorkItem. But frameworks can derive their own context from SynchronizationContext and override the Post method to be more appropriate to the scheduler being represented. In the case of Windows Forms, for example, the WindowsFormsSynchronizationContext implements Post to pass the delegate off to Control.BeginInvoke. For DispatcherSynchronizationContext in WPF, it calls to Dispatcher.BeginInvoke. And so on.

That’s how await “brings you back to where you were.” It asks for the SynchronizationContext that’s representing the current environment, and then when the await completes, the continuation is posted back to that context. It’s up to the implementation of the captured context to run the delegate in the right place, e.g. in the case of a UI app, that means running the delegate on the UI thread. This explanation also helps to highlight what happens if the environment didn’t set a SynchronizationContext onto the current thread (and if there’s not special TaskScheduler, as there isn’t in this case). If the context comes back as null, then the continuation could run “anywhere”. I put anywhere in quotes because obviously the continuation can’t run “anywhere,” but logically you can think of it like that… it’ll either end up running on the same thread that completed the awaited task, or it’ll end up running in the ThreadPool.

All of the UI application types you can create in Visual Studio will end up having a special SynchronizationContext published on the UI thread. Windows Forms, Windows Presentation Foundation, Metro style apps… they all have one. But there’s one common kind of application that doesn’t have a SynchronizationContext: console apps. When your console application’s Main method is invoked, SynchronizationContext.Current will return null. That means that if you invoke an asynchronous method in your console app, unless you do something special, your asynchronous methods will not have thread affinity: the continuations within those asynchronous methods could end up running “anywhere.”

As an example, consider this application:

using System;

using System.Collections.Generic;

using System.Threading;

using System.Threading.Tasks;

class Program

{

static void Main()

{

DemoAsync().Wait();

}

static async Task DemoAsync()

{

var d = new Dictionary<int, int>();

for (int i = 0; i < 10000; i++)

{

int id = Thread.CurrentThread.ManagedThreadId;

int count;

d[id] = d.TryGetValue(id, out count) ? count+1 : 1;

await Task.Yield();

}

foreach (var pair in d) Console.WriteLine(pair);

}

Here I’ve created a dictionary that maps thread IDs to the number of times we encountered that particular thread. For thousands of iterations, I get the current thread’s ID and increment the appropriate element of my histogram, then yield. The act of yielding will use a continuation to run the remainder of the method. Here’s some representative output I see from executing this app:

[1, 1]

[3, 2687]

[4, 2399]

[5, 2397]

[6, 2516]

Press any key to continue . . .

We can see here that the execution of this code used 5 threads over the course of its run. Interestingly, one of the threads only had one hit. Can you guess which thread that was? It’s the thread running the Main method of the console app. When we call DemoAsync, it runs synchronously until the first await the yields, so the first time we check the ManagedThreadId for the current thread, we’re still on the thread that invoked DemoAsync. Once we hit the await, the method returns back to Main(), which then blocks waiting on the returned Task to complete. The continuations used by the remainder of the async method’s execution would have been posted to SynchronizationContext.Current, except that it a console app, it’s null (unless you explicitly override that with SynchronizationContext.SetSynchronizationContext). So the continuations just get scheduled to run on the ThreadPool. That’s where the rest of those threads are coming from… they’re all ThreadPool threads.

Is it a problem then that using async like this in a console app might end up running continuations on ThreadPool threads? I can’t answer that, because the answer is entirely up to what kind of semantics you need in your application. For many applications, this will be perfectly reasonable behavior. Other applications, however, may require thread affinity, such that all of the continuations run on the same thread. For example, if you invoked multiple async methods concurrently, you might want all the continuations they use to be serialized, and an easy way to guarantee that is to ensure that only one thread is used for executing all of the continuations. If your application does demand such behavior, are you out of luck? Thankfully, the answer is ‘no’. You can add such behavior yourself.

If you’ve made it this far in reading, hopefully the components of a solution here have started to become obvious. You effectively need a message pump, a scheduler, something that runs on the Main thread of your app processing a queue of work. And you need a SynchronizationContext (or a TaskScheduler if you prefer) that feeds the await continuations into that queue. With that framework in place, let’s build a solution.

First, we need our SynchronizationContext. As described in the previous paragraph, we’ll need a queue to store the work to be done. The work provided to the Post method comes in the form of two objects: a SendOrPostCallback delegate, and an object state that is meant to be passed into that delegate when it’s invoked. As such, we’ll have our queue store a KeyValuePair<TKey,TValue> of these two objects. What kind of queue data structure should we use? We need something ideally suited to handle producer/consumer scenarios, as our asynchronous method will be “producing” these pairs of work, and our pumping loop will need to be “consuming” them from the queue and executing them. .NET 4 saw the introduction of the perfect type for the job: BlockingCollection<T>. BlockingCollection<T> is a data structure that encapsulates not only a queue, but also all of the synchronization necessary to coordinate between a producer adding elements to that queue and a consumer removing them, including blocking the consumer attempting a removal while the queue is empty.

With that, the pieces fall into place: a BlockingCollection<KeyValuePair<SendOrPostCallback,object>> instance; a Post method that adds to the queue; another method that sits in a consuming loop, removing each work item and processing it; and finally another method that lets the queue know that no more work will arrive, allowing the consuming loop to exit once the queue is empty.

private sealed class SingleThreadSynchronizationContext :

SynchronizationContext

{

private readonly

BlockingCollection<KeyValuePair<SendOrPostCallback,object>>

m_queue =

new BlockingCollection<KeyValuePair<SendOrPostCallback,object>>();

public override void Post(SendOrPostCallback d, object state)

{

m_queue.Add(

new KeyValuePair<SendOrPostCallback,object>(d, state));

}

public void RunOnCurrentThread()

{

KeyValuePair<SendOrPostCallback, object> workItem;

while(m_queue.TryTake(out workItem, Timeout.Infinite))

workItem.Key(workItem.Value);

}

public void Complete() { m_queue.CompleteAdding(); }

…

}

Believe it or not, we’re already half done with our solution. We need to instantiate one of these contexts and set it as current onto the current thread, so that when we then invoke the asynchronous method, that method’s awaits will see this context as Current. We need to alert the context to when there won’t be any more work arriving, which we can do by using a continuation to call Complete on our context when the Task returned from the async method is compelted. We need to run the processing loop via the context’s RunOnCurrentThread method. And we need to propagate any exceptions that may have occurred during the async method’s processing. All in all, it’s just a few lines:

public static void Run(Func<Task> func)

{

var prevCtx = SynchronizationContext.Current;

try

{

var syncCtx = new SingleThreadSynchronizationContext();

SynchronizationContext.SetSynchronizationContext(syncCtx);

var t = func();

t.ContinueWith(

delegate { syncCtx.Complete(); }, TaskScheduler.Default);

syncCtx.RunOnCurrentThread();

t.GetAwaiter().GetResult();

}

finally { SynchronizationContext.SetSynchronizationContext(prevCtx); }

}

That’s it. With our solution now available, I can change the Main method of my demo console app from:

static void Main()

{

DemoAsync().Wait();

}

to instead use our new AsyncPump.Run method:

static void Main()

{

AsyncPump.Run(async delegate

{

await DemoAsync();

});

}

When I then run my app again, this time I get the following output:

[1, 10000]

Press any key to continue . . .

As you can see, all of the continuations have run on just one thread, the main thread of my console app.

The AsyncPump sample class described in this post is available as an attachment to this post.

FAQ on Task.Start

Stephen Toub - MSFT — Sun, 15 Jan 2012 02:03:53 GMT

Recently I’ve heard a number of folks asking about Task.Start, when and when not to use it, how it behaves,and so forth. I thought I’d answer some of those questions here in an attempt to clarify and put to rest any misconceptions about what it is and what it does.

1. Question: When can I use Task.Start?

The Start instance method may be used if and only if the Task is in the Created state (i.e. Task.Status returns TaskStatus.Created). And the only way a Task can be in the Created state is if the Task were instantiated using one of Task’s public constructors, e.g. "var t = new Task(someDelegate);”.

2. Question: Should I call Start on a Task created by Task.Run / Task.ContinueWith / Task.Factory.StartNew / TaskCompletionSource / async methods / …?

No. Not only shouldn’t you, but you simply can’t… it would fail with an exception. See question #1. The Start method is only applicable to a Task in the Created state. Tasks created by all of those mentioned means are already beyond the Created state, such that their Task.Status will not return TaskStatus.Created, but something else, like TaskStatus.WaitingForActivation, or TaskStatus.Running, or TaskStatus.RanToCompletion.

3. Question: What does Start actually do?

It queues the Task to the target TaskScheduler (the parameterless overload of Start targets TaskScheduler.Current). When you construct a Task with one of Task’s constructors, the Task is inactive: it has not been given to any scheduler yet, and thus there’s nothing to actually execute it. If you never Start a Task, it’ll never be queued, and so it’ll never complete. To get the Task to execute, it needs to be queued to a scheduler, so that the scheduler can execute it when and where the scheduler sees fit to do so. The act of calling Start on a Task will twiddle some bits in the Task (e.g. changing its state from Created to WaitingToRun) and will then pass the Task to the target scheduler via the TaskScheduler’s QueueTask method. At that point, the task’s future execution is in the hands of the scheduler, which should eventually execute the Task via the TaskScheduler’s TryExecuteTask method.

4. Question: Can I call Start more than once on the same Task?

No. A Task may only transition out of the Created state once, and Start transitions a Task out of the Created state: therefore, Start may only be used once. Any attempts to call Start on a Task not in the Created state will result in an exception. The Start method employs synchronization to ensure that the Task object remains in a consistent state even if Start is called multiple times concurrently… only one of those calls may succeed.

5. Question: What’s the difference between using Task.Start and Task.Factory.StartNew?

Task.Factory.StartNew is shorthand for new’ing up a Task and Start’ing it. So, the following code:

var t = Task.Factory.StartNew(someDelegate);

is functionally equivalent to:

var t = new Task(someDelegate);
t.Start();

Performance-wise, the former is slightly more efficient. As mentioned in response to question #3, Start employs synchronization to ensure that the Task instance on which Start is being called hasn’t already been started, or isn’t concurrently being started. In contrast, the implementation of StartNew knows that no one else could be starting the Task concurrently, as it hasn’t given out that reference to anyone… so StartNew doesn’t need to employ that synchronization.

6. Question: I’ve heard that Task.Result may also start the Task. True?

False. There are only two ways that a Task in the Created state may transition out of that state:

A CancellationToken was passed into the Task’s constructor, and that token then had or then has cancellation requested. If the Task is still in the Created state when that happens, it would transition into the Canceled state.
Start is called on the Task.

That’s it, and notice that Result is not one of those two. If you use .Wait() or .Result on a Task in the Created state, the call will block; someone else would need to Start the Task so that it could then be queued to a scheduler, so that the scheduler could eventually execute it, and so that the Task could complete… the blocking call could then complete as well and wake up.

What you might be thinking of isn’t that .Result could start the task, but that it could potentially “inline” the task’s execution. If a Task has already been queued to a TaskScheduler, then that Task might still be sitting in whatever data structure the scheduler is using to store queued tasks. When you call .Result on a Task that’s been queued, the runtime can attempt to inline the Task’s execution (meaning to run the Task on the calling thread) rather than purely blocking and waiting for some other thread used by the scheduler to execute the Task at some time in the future. To do this, the call to .Result may end up calling the TaskScheduler’s TryExecuteTaskInline method, and it’s up to the TaskScheduler how it wants to handle the request.

7. Question: Should I return unstarted Tasks from public APIs?

The proper question is “Should I return Tasks in the Created state from public APIs?” And the answer is “No.” (I draw the distinction in the question here due to questions #1 and #2 above… the majority of mechanisms for creating a Task don’t permit for Start to be called, and I don’t want folks to get the impression that you must call Start on a Task in order to allow it to be returned from a public API… that is not the case.)

The fundamental idea here is this. When you call a normal synchronous method, the invocation of that method begins as soon as you’ve invoked it. For a method that returns a Task, you can think of that Task as representing the eventual asynchronous completion of the method. But that doesn’t change the fact that invoking the method begins the relevant operation. Therefore, it would be quite odd if the Task returned from the method was in the Created state, which would mean it represents an operation that hasn’t yet begun.

So, if you have a public method that returns a Task, and if you create that Task using one of Task’s constructors, make sure you Start the Task before returning it. Otherwise, you’re likely to cause a deadlock or similar problem in the consuming application, as the consumer will expect the Task to eventually complete when the launched operation completes, and yet if such a Task hasn’t been started, it will never complete. Some frameworks that allow you to parameterize the framework with methods/delegates that return Tasks even validate the returned Task’s Status, throwing an exception if the Task is still Created.

8. Question: So, should I use Task’s ctor and Task.Start?

In the majority of cases, you’re better off using some other mechanism. For example, if all you want to do is schedule a Task to run some delegate for you, you’re better off using Task.Run or Task.Factory.StartNew, rather than constructing the Task and then Start’ing it; not only will the former methods result in less code, but they’re also cheaper (see question #5 above), and you’re less likely to make a mistake with them, such as forgetting to Start the Task.

There are of course valid situations in which using the ctor + Start makes sense. For example, if you choose to derive from Task for some reason, then you’d need to use the Start method to actually queue it. A more advanced example is if you want the Task to get a reference to itself. Consider the following (buggy) code:

Task theTask = null;
theTask = Task.Run(() => Console.WriteLine(“My ID is {0}.”, theTask.Id));

Spot the flaw? There’s a race. During the call to Task.Run, a new Task object is created and is queued to the ThreadPool scheduler. If there’s not that much going on in the ThreadPool, a thread from the pool might pick it up almost instantly and start running it. That thread is now racing to access the variable ‘theTask’ with the main thread that called Task.Run and that needs to store the created Task into that ‘theTask’ variable. I can fix this race by separating the construction and scheduling:

Task theTask = null;
theTask = new Task(() =>Console.WriteLine(“My ID is {0}.”, theTask.Id));
theTask.Start(TaskScheduler.Default);

Now I’m now sure that the Task instance will have been stored into the ‘theTask’ variable before the ThreadPool processes the Task, because the ThreadPool won’t even get a reference to the Task until Start is called to queue it, and by that point, the reference has already been set (and for those of you familiar with memory models, the appropriate fences are put in place by Task to ensure this is safe).

Awaiting Socket Operations

Stephen Toub - MSFT — Thu, 15 Dec 2011 22:36:00 GMT

The System.Net.Sockets.Socket class in .NET exposes multiple sets of asynchronous methods that perform the same basic operations but that are exposed with different patterns.

The first set follows the APM pattern, where for a synchronous method like Receive, the BeginReceive and EndReceive methods are exposed. If you want to be able to “await” such asynchronous operations in an async method, the easiest way to do so is to create a Task-based wrapper. That can be done using Task.Factory.FromAsync, or it can be done more manually by using a TaskCompletionSource<TResult>, e.g.

public static Task<int> ReceiveAsync(
    this Socket socket, byte[] buffer, int offset, int size,
    SocketFlags socketFlags)
{
    var tcs = new TaskCompletionSource<int>();
    socket.BeginReceive(buffer, offset, size, socketFlags, iar =>
    {
        try { tcs.TrySetResult(socket.EndReceive(iar)); }
        catch (Exception exc) { tcs.TrySetException(exc); }
    }, null);
    return tcs.Task;
}

or with a slightly more efficient implementation that avoids some of the extra allocations here:

public static Task<int> ReceiveAsync(
    this Socket socket, byte[] buffer, int offset, int size,
    SocketFlags socketFlags)
{
    var tcs = new TaskCompletionSource<int>(socket);
    socket.BeginReceive(buffer, offset, size, socketFlags, iar =>
    {
        var t = (TaskCompletionSource<int>)iar.AsyncState;
        var s = (Socket)t.Task.AsyncState;
        try { t.TrySetResult(s.EndReceive(iar)); }
        catch (Exception exc) { t.TrySetException(exc); }
    }, tcs);
    return tcs.Task;
}

For many asynchronous methods, this approach is good enough. The overhead involved in creating a new TaskCompletionSource<TResult> for each operation is often dwarfed by the costs of the operation itself.

However, there are some scenarios where this is not sufficient, and not only the additional tasks, but the APM implementation itself. If you’re making thousands upon thousands of socket calls asynchronously per second, that’s thousands upon thousands of IAsyncResult objects getting created. That’s a lot of pressure on the garbage collector, and can result in unacceptable pauses in some networking-intensive apps.

To address that, Socket exposes another set of asynchronous methods. These methods look similar to the Event-based Async Pattern (EAP), but they’re subtly different. Basically, you create a SocketAsyncEventArgs instance, and you configure that instance with a buffer, with a Completion event handler, and so on. Then you pass this instance into methods like ReceiveAsync and SendAsync. Those methods return a Boolean indicating whether they completed synchronously, and if they did, then the event args instance is immediately reusable, and the caller is responsible for finishing the operation synchronously (e.g. checking the status of the operation, getting from the SocketAsyncEventArgs instance the number of bytes received, etc.) If the operation didn’t complete synchronously, then the completion event will be raised when it does complete, and that handler does the same completion logic we would have done synchronously. While a bit more complicated, the benefit of this pattern is that these SocketAsyncEventArgs instances can be reused, such that once an instance is no longer being used by a first asynchronous call, it may be reused for a second, and then for a third, and so on. This avoids needing to construct additional objects per async operation, which thereby decreases pressure on the garbage collector.

We could of course wrap such asynchronous methods with Task, as well, but that would largely defeat the purpose of these methods, similar to allocating an IAsyncResult in the APM case. Still, though, we want to be able to take advantage of the compiler’s async/await support to make it easier to write asynchronous code using sockets. We want our cake and to eat it, too.

Of course, the compiler supports awaiting more than just Tasks. So if you have a specialized scenario like this, you can take advantage of the compiler’s pattern-based support for awaiting things. Below, I’ve implemented an awaitable wrapper for a SocketAsyncEventArgs:

public sealed class SocketAwaitable
{
    private readonly static Action SENTINEL = () => { };

    internal bool m_wasCompleted;
    internal Action m_continuation;
    internal SocketAsyncEventArgs m_eventArgs;

    public SocketAwaitable(SocketAsyncEventArgs eventArgs)
    {
        if (eventArgs == null) throw new ArgumentNullException("eventArgs");
        m_eventArgs = eventArgs;
        eventArgs.Completed += delegate
        {
            var prev = m_continuation ?? Interlocked.CompareExchange(
                ref m_continuation, SENTINEL, null);
            if (prev != null) prev();
        };
    }

    internal void Reset()
    {
        m_wasCompleted = false;
        m_continuation = null;
    }

    public SocketAwaitable GetAwaiter() { return this; }

    public bool IsCompleted { get { return m_wasCompleted; } }

    public void OnCompleted(Action continuation)
    {
        if (m_continuation == SENTINEL ||
            Interlocked.CompareExchange(
                ref m_continuation, continuation, null) == SENTINEL)
        {
            Task.Run(continuation);
        }
    }

    public void GetResult()
    {
        if (m_eventArgs.SocketError != SocketError.Success)
            throw new SocketException((int)m_eventArgs.SocketError);
    }
}

With that helper type, I can write a short three-line wrapper method for each of the Socket methods I care about, e.g.

public static class SocketExtensions
{
    public static SocketAwaitable ReceiveAsync(this Socket socket,
        SocketAwaitable awaitable)
    {
        awaitable.Reset();
        if (!socket.ReceiveAsync(awaitable.m_eventArgs))
            awaitable.m_wasCompleted = true;
        return awaitable;
    }

    public static SocketAwaitable SendAsync(this Socket socket,
        SocketAwaitable awaitable)
    {
        awaitable.Reset();
        if (!socket.SendAsync(awaitable.m_eventArgs))
            awaitable.m_wasCompleted = true;
        return awaitable;
    }

    // ...
}

And with that, I can use async/await with these very efficient asynchronous methods on Socket, e.g.

static async Task ReadAsync(Socket s)
{
    // Reusable SocketAsyncEventArgs and awaitable wrapper
    var args = new SocketAsyncEventArgs();
    args.SetBuffer(new byte[0x1000], 0, 0x1000);
    var awaitable = new SocketAwaitable(args);

    // Do processing, continually receiving from the socket
    while (true)
    {
        await s.ReceiveAsync(awaitable);
        int bytesRead = args.BytesTransferred;
        if (bytesRead <= 0) break;

        Console.WriteLine(bytesRead);
    }
}

Here, when I make a call to my ReceiveAsync extension method, I’m using a reusable SocketAsyncEventArgs that was previously created and wrapped into one of my helper SocketAwaitable types. ReceiveAsync doesn’t need to do any allocation; it simply calls to the underlying ReceiveAsync on the Socket, passing in the wrapped SocketAsyncEventArgs. If the operation completes synchronously, we just set the awaitable’s m_wasCompleted to true, such that when the compiler-generated code awaits the returned SocketAwaitable, it sees that IsCompleted is true, and simply proceeds to call GetResult, which will throw an exception if the socket operation failed. If the underlying ReceiveAsync call does not complete synchronously, then m_wasCompleted will remain false, and it’ll be up to the OnCompleted method to store the await continuation to be run when the SocketAsyncEventArgs’ Completed handler is invoked. The only time we need to schedule a task here is if the socket operation asynchronously completes between the time that the compiler-generated code checks IsCompleted and then invokes OnCompleted, and that should be rare.

If you wanted to write a bit more code, you could even create a specialized version of SocketAwaitable such that its GetResult method returned the result of the operation, e.g. a ReceiveSocketAwaitable whose GetResult method returned the SocketAsyncEventArgs.BytesTransferred value. With that, you could then write the above loop almost as if it were synchronous code, e.g.

static async Task ReadAsync(Socket s)
{
    // Reusable SocketAsyncEventArgs and awaitable wrapper
    var args = new SocketAsyncEventArgs();
    args.SetBuffer(new byte[0x1000], 0, 0x1000);
    var awaitable = new ReceiveSocketAwaitable(args);

    // Do processing, continually receiving from the socket
    int bytesRead;
    while ((bytesRead = await s.ReceiveAsync(awaitable)) > 0)
    {
        Console.WriteLine(bytesRead);
    }
}

Enjoy.

Paper :: Guide to Implementing Custom TPL Dataflow Blocks

Danny Shih — Mon, 05 Dec 2011 18:14:00 GMT

TPL Dataflow includes a number of built-in, already-implemented blocks that target the most common scenarios. Additionally, some flexibility is provided by the set of options that may be used to tweak block behaviors. However, a developer may still choose to implement a custom block for advanced scenarios where the built-in ones are not sufficient. Zlatko Michailov explores the options for achieving this in his paper:

Guide to Implementing Custom TPL Dataflow Blocks

Enjoy!

Coalescing CancellationTokens from Timeouts

Stephen Toub - MSFT — Sun, 04 Dec 2011 02:58:00 GMT

In the .NET Framework 4.5 Developer Preview, you’ll find that CancellationTokenSource now has timeout support built directly into its implementation. This makes it very easy to create a token that will automatically have cancellation requested after a particular time interval, e.g.

public static CancellationToken FromTimeout(int millisecondsDue)
{
return new CancellationTokenSource(millisecondsDue).Token;
}

Under the covers here, CancellationTokenSource is creating and configuring a new System.Threading.Timer to call the source’s Cancel method after the particular duration passes. The CLR team did some great work for .NET 4.5 to significantly improve the performance of Timer over .NET 4, making this approach not only easy but also well-performing. Of course, there are always scenarios the push the boundaries of performance, and we’ve recently seen some high-throughput cases where folks were creating one such CancellationToken for each of thousands upon thousands of asynchronous calls being made per second. That’s a lot of Timer and CancellationTokenSource instances. Even with that, the overhead was reasonable, but we’re constantly looking for ways to decrease costs.

One such solution for the consumer is to coalesce tokens. Rather than using a unique CancellationToken per request, we can create one CancellationToken for all timeout requests that will expire at approximately the same time (where we get to configure the size of the span that defines “approximately the same time”). That way, we can pick how much flexibility we’re willing to accept in the timeout (there already needs to be some flexibility anyway given what the OS can guarantee, what other work is on the system consuming cycles, etc.), and trade that off against resource utilization. Here’s one way this can be implemented, here using a 50 millisecond coalescing span:

public static class CoalescedTokens

{

    private const uint COALESCING_SPAN_MS = 50;

    private readonly static
        ConcurrentDictionary<long, CancellationToken> s_timeToToken =
        new ConcurrentDictionary<long, CancellationToken>();

    public static CancellationToken FromTimeout(int millisecondsTimeout)

    {

        if (millisecondsTimeout <= 0)

            return new CancellationToken(true);

        uint currentTime = (uint)Environment.TickCount;

        long targetTime = millisecondsTimeout + currentTime;

        targetTime = ((targetTime + (COALESCING_SPAN_MS - 1))

            / COALESCING_SPAN_MS) * COALESCING_SPAN_MS;

        CancellationToken token;

        if (!s_timeToToken.TryGetValue(targetTime, out token))
        {

            token = new CancellationTokenSource(
                (int)(targetTime - currentTime).Token;

            if (s_timeToToken.TryAdd(targetTime, token))

            {

                token.Register(state => {

                    CancellationToken ignored;

                    s_timeToToken.TryRemove((long)state, out ignored);

                }, targetTime);

            }
        }

        return token;

    }

}

With that, we can now use CoalescedTokens.FromTimeout to get a CancellationToken that will be canceled within approximately 50 milliseconds of the timeout I requested, such that this same token will be handed out to anyone else asking for a timeout that would expire in the same span. The tokens are all stored in a ConcurrentDictionary, indexed by the time at which the token will have cancellation requested. We first figure out what coalescing interval the incoming request maps to, and look that up in the dictionary. If the token exists, we return it. If it doesn’t, we create a new token that will have cancellation requested at the right time, try to add it to the dictionary, and return it. If we were successful at adding it to the dictionary, we also register a cancellation callback that will remove it from the dictionary once cancellation is requested, avoiding a potential leak. Of course, there’s a chance there’s a race where two requests come in concurrently, don’t find an existing token, and each create their own… and that’s fine. It’s a “benign” race, in that each will just get its own token that will expire at the right time, though only one of them will be published for subsequent requestors to use.

For all but the most intensive of scenarios, if you need a token that will have cancellation requested after some timeout, just do the simple thing of using “new CancellationTokenSource(timeout).Token” and all should be well. But, if after careful profiling and measurement, you find that this is a bottleneck, hopefully this approach and example can help you to address it… one more bit of code for your bag of tricks.

How to use C++ AMP from C# using WinRT

Igor Ostrovsky - MSFT — Sat, 12 Nov 2011 00:01:00 GMT

In a previous article, How to use C++ AMP to C#, we described how you can use P/Invoke to call into C++ AMP and accelerate your C# apps on GPUs and other heterogeneous hardware. In this post, we'll take a look at how the same task becomes easier in Windows 8 using WinRT.

Before attempting to call C++ AMP from C#, make sure that you have C++ AMP working on your machine. Daniel Moth’s blog post VS 11 Developer Preview gotchas with C++ AMP explains how to get that done.

The short story

Once you have C++ AMP working on your machine, the easiest way to start using it from C# via WinRT is to open this sample project in Visual Studio 11 and begin experimenting with the code.

The long story

If you have an existing Metro style app that you’d like to modify to use C++ AMP – or you’d like to understand how the sample is set up – you can follow the steps below. In summary, you need to take the following steps:

Step 1: Open or create a C# Metro style project in Visual Studio 11

Choose the platform target as X86 (if you plan to write 32-bit C++ AMP code).

Step 2: Create a C++ WinRT Component DLL project

DO NOT build the project before completing step 3

Step 3: Add the C++ project as a reference to the C# project.
Step 4: Write the C++ AMP and the C# code

Step 1: Open or create a C# Metro style project

First, you need to open or create a C# Metro style application project. The rest of the article assumes that the project is named HelloWorldCSharpWinRT:

Also, set the "Platform target" of the project to "X86".

Step 2: Create a C++ WinRT Component DLL project

Now, you can add a Visual C++ WinRT component that will contain the C++ AMP code. Simply create a project named "HelloWorldLib" from the "WinRT Component DLL" template:

WARNING: Do not build the project yet! Due to a bug in the Visual Studio 11 Developer Preview, building the project before completing step 3 will cause problems, and you may be stuck having to delete and recreate the C++ WinRT project.

Step 3: Add reference from HelloWorldCSharpWinRT to HelloWorldLib

With WinRT, you can simply add HelloWorldLib as a reference to HelloWorldCSharpWinRT. No more manual editing of the csproj file is necessary, as it was with P/Invoke! Just right-click HelloWorldCSharpWinRT, choose "Add Reference..." and select the HelloWorldLib project:

Step 4. Write the C++ AMP and the C# code

Now, we just need to write the C++ AMP code and call it from C#.

Since a C++ AMP kernel may take a long time to execute, the WinRT guidelines state that the kernel should be exposed as an asynchronous operation. A convenient way to expose asynchronous operations in C++ is via create_async, currently available in the PPL Sample Pack (for details on how this works, see Try It Now: Use PPL to Produce Windows 8 Asynchronous Operations).

Extract ppltasks.preview.h from the sample pack and include it as a header file in HelloWorldCSharpWinRT.

Delete WinRTComponent.h.

Modify WinRTComponent.cpp as follows:

#include "amp.h"
#include "ppltasks.preview.h"
#include "collection.h"
#include <vector>

using namespace concurrency;
using namespace Windows::Foundation;
using namespace Windows::Foundation::Collections;

namespace HelloWorldLib
{
    public ref class WinRTComponent sealed
    {
    public:
        IAsyncOperation<IVectorView<float>^>^ square_array_async(
            IVectorView<float>^ input)
        {
            // Synchronously copy input data from host to device
            int size = input->Size;
            array<float, 1> *dataPt = new array<float, 1>(
                size, begin(input), end(input));

            // Asynchronously perform the computation on the GPU
            return preview::create_async( [=]() -> IVectorView<float>^
            {
                // Run the kernel on the GPU
                parallel_for_each(
                    dataPt->grid, [=] (index<1> idx) mutable restrict(direct3d)
                {
                    (*dataPt)[idx] = (*dataPt)[idx] * (*dataPt)[idx];
                });

                // Copy outputs from device to host
                std::vector<float> vec = std::vector<float>(size);
                copy((*dataPt), vec.begin());
                delete dataPt;

                // Return the outputs as a VectorView<float>
                return ref new Platform::VectorView<float>(vec);
            });
        }
    };
}

Notice that the square-array operation is exposed via an asynchronous API. In WinRT, operations that may be long-running should be exposed via asynchronous APIs, and GPU operations may take a relatively long time to execute.

That is all that we need on the C++ side. Now, we'll add a button to the C# project. Modify MainPage.xaml as follows:

When the user clicks the button, we'll call into C++ AMP. Modify MainPage.xaml.cs as follows:

using System;
using System.Collections.Generic;
using Windows.UI.Popups;
using Windows.UI.Xaml;
using HelloWorldLib;

namespace HelloWorldCSharpWinRT
{
    partial class MainPage
    {
        public MainPage()
        {
            InitializeComponent();
        }

        private async void Button_Example_Click(
            object sender, RoutedEventArgs e)
        {
            Button_Example.IsEnabled = false;
            var arr = new [] { 1.0f, 2.0f, 3.0f, 4.0f };
            List<float> inputs = new List<float>(arr);

            IReadOnlyList<float> outputs =
                await new WinRTComponent()
                    .square_array_async(inputs);

            await new MessageDialog(string.Join(",", outputs)).ShowAsync();
            Button_Example.IsEnabled = true;
        }
    }
}

… and that’s it!

Note that this is a very simple example that demonstrates how to call a C++ AMP function from C#. The example is too naïve to demonstrate speedup – it contains too little work per data element and in total to benefit from GPU acceleration. An example of a workload that does demonstrate speedup is matrix multiplication, and here is a link to C++ AMP code for Matrix Multiplication.

New in .NET 4.5: ThreadLocal.Values

Igor Ostrovsky - MSFT — Fri, 11 Nov 2011 01:12:50 GMT

Available since .NET 4, ThreadLocal<T> is a container that holds a separate value for every thread. In practice, ThreadLocal<T> is often convenient for storing per-thread counters, resources, or partial results.

As mentioned earlier on this blog, we have been thinking about adding a Values property to enumerate over the values from all threads that ever stored a value into the ThreadLocal<T> instance. After getting helpful feedback from this blog and other sources, we ended up implementing the capability in .NET 4.5.

To use the Values property, you have to initialize the ThreadLocal<T> instance by setting the constructor argument trackAllValues= true. The sample below demonstrates this usage:

var localResult = new ThreadLocal<int>(() => 0, trackAllValues: true);
Parallel.For(0, 10000, i =>
{
    localResult.Value += Compute(i);
});

int result = localResult.Values.Sum();

Explicit Opt-In

If you don’t set the trackAllValues constructor argument to true, accessing the Values property will throw an InvalidOperationException. The Values property requires additional bookkeeping, and so ThreadLocal<T> requires that you declare upfront if you plan to access Values.

Most importantly, trackAllValues=true changes the lifetime of the values stored into a ThreadLocal<T>. By default (i.e., trackAllValues=false), a value stored into a ThreadLocal<T> is cleared when the ThreadLocal<T> instance is disposed/finalized OR when the corresponding thread exits, whichever happens first. This is the same behavior that ThreadLocal<T> had in .NET 4, and can be especially important when T is a large reference type.

However, removing a value when its corresponding thread exits would make the Values property poorly behaved. For example, in the code sample shown earlier in this blog post, the final summation should clearly process all partial results. It would be unacceptable to omit one of the partial results, say because the ThreadPool has decided to retire the thread that computed it.

So, if you set trackAllValues=true, the ThreadLocal instance will keep track of all values that have been stored into it, even if the corresponding threads are now gone. Then, the values can be returned via the Values property.

More Scenarios

In addition to aggregation of partial results, the Values property can be useful for resource cleanup. The example below executes a Parallel.For loop, where each thread participating in the loop opens its own connection to the database. Once the loop completes, we can use the Values property to release all connections that have been opened:

var threadDbConn = new ThreadLocal<MyDbConnection>(() => MyDbConnection.Open(), true);
try
{
    Parallel.For(0, 10000, i =>
    {
        var inputData = threadDbConn.Value.GetData(i);
        ...
    });
}
finally
{
    foreach(var dbConn in threadDbConn.Values)
    {
        dbConn.Close();
    }
}

Additionally, readers of this blog suggested in comments that the Values property would be useful to simplify existing parallel loops, to aggregate data parallel code that does not use parallel loops, to implement an optimized object pool, and to gather high-performance statistics.

If you find this new property useful in your code, please let us know in the comments!

PLINQ Queries That Run in Parallel in .NET 4.5

Igor Ostrovsky - MSFT — Fri, 11 Nov 2011 01:12:00 GMT

One interesting thing to know about PLINQ is that not all queries are guaranteed to execute in parallel (See PLINQ Queries That Run Sequentially for reference). You can think of the AsParallel method as a hint to run in parallel for query shapes that it believes will be faster. By default, PLINQ prefers to use a simple sequential algorithm over a potentially expensive parallel algorithm, and so some queries will execute sequentially, especially queries where the chain of operators do not lend themselves to parallelism. The bottom line is that PLINQ’s intent is to run your queries as fast as possible, so if there’s a good likelihood that the complexities of parallelism will slow it down, PLINQ will perform the query sequentially rather than in parallel.

In .NET 4.5, PLINQ’s algorithms have improved over .NET 4 such that its set of parallelizable queries has broadened and substantially fewer queries fall back to sequential execution. For example, the following query would execute sequentially in .NET 4 but now executes in parallel in .NET 4.5:

intArray.AsParallel()
    .Select(x => Foo(x))
    .TakeWhile(x => Filter(x))
    .ToArray();

Since the query executes sequentially in .NET 4, the query would not get any speedup even if the Foo() method is expensive. For some cases, the choice to fallback to a sequential implementation might be too conservative, and as a developer you might choose to override it. In order to force PLINQ to use a parallel algorithm in .NET 4, the user can call the WithExecutionMode operator:

intArray.AsParallel()
    .WithExecutionMode(ParallelExecutionMode.ForceParallelism)
    .Select(x => Foo(x))
    .TakeWhile(x => Filter(x))
    .ToArray();

In .NET 4.5, the behavior of the TakeWhile operator has changed, and so the query executes in parallel even without the WithExecutionMode hint.

Sequential Fallback in .NET 4 and .NET 4.5

One way to illustrate the improvements to the sequential fallback is to look at the list of operators that may cause the query to fallback to sequential. Whether or not the operator actually causes a sequential fallback depends on other operators in the query as well.

Operators that may cause sequential fallback in both .NET 4 and .NET 4.5 are marked in blue, and operators that may cause fallback in .NET 4 but no longer in .NET 4.5 are marked in orange.

So, in .NET 4.5, as long as the query does not contain positional operators or the ElementAt operator, it will not fall back to sequential.

The positional operators mentioned above are the operator overloads that pass the position of each element into the user delegate. So, while the ‘ordinary’ Select operator accepts a delegate of type Func<TInput, TOutput>, the positional Select accepts a delegate of type Func<TInput, int, TOutput>. The position of each element is passed into the delegate as the input second parameter.

The ForceParallelism execution mode is still available for those items called out above.

Potential Risks of the Change

The fact that some queries used to execute sequentially and now execute in parallel opens up the possibility of exposing more latent race conditions in users’ code.

Here is an example of a query that happens to work in .NET 4, but will not work in .NET 4.5:

List<int> list = new List<int>();
var q = src.AsParallel()
    .Select(x => { list.Add(x); return x; })
    .Where(x => true)
    .Take(100);

There is a bug in the query above: the delegate passed into the Select operator is not thread-safe. However, the query will still happen to work in .NET 4, because as an implementation detail PLINQ decides to execute the query sequentially. In .NET 4.5, the query will run in parallel, and so list.Add(x) could be called from multiple threads concurrently, potentially resulting in a corrupted List<T> instance.

However, the idea that user delegates may be called from multiple cores is a core premise of PLINQ, so hopefully no user code in the real world will be broken by the change.

Related Resources:

PLINQ Queries That Run Sequentially
MSDN Documentation for ParallelExecutionMode Enumeration
How to: Specify the Execution Mode in PLINQ
Understanding Speedup in PLINQ

Paper :: TPL Performance Improvements in .NET 4.5

Danny Shih — Thu, 10 Nov 2011 22:39:25 GMT

We invested a lot of time into making parallel programming “just faster” for .NET 4.5. You’ve already seen some neat tricks to ConcurrentDictionary. There’s a lot more to say about improving the performance of the Task Parallel Library, and Joe Hoag has a written an excellent paper on the subject:

TPL Performance Improvements in .NET 4.5

The paper examines some of the changes made to TPL in .NET 4.5, compares times for selected operations in .NET 4 vs .NET 4.5, and makes some general “best practices” recommendations for those interested in using TPL performantly in .NET 4.5.

Enjoy!

Crafting a Task.TimeoutAfter Method

Joe Hoag — Thu, 10 Nov 2011 15:50:11 GMT

Imagine that you have a Task handed to you by a third party, and that you would like to force this Task to complete within a specified time period. However, you cannot alter the “natural” completion path and completion state of the Task, as that may cause problems with other consumers of the Task. So you need a way to obtain a copy or “proxy” of the Task that will either (A) complete within the specified time period, or (B) will complete with an indication that it had timed out.

In this blog post, I will show how one might go about implementing a Task.TimeoutAfter method to support this scenario. The signature of TimeoutAfter would look like this:

public static Task TimeoutAfter(this Task task, int millisecondsTimeout)

The returned proxy Task can complete in one of two ways:

If task completes before the specified timeout period has elapsed, then the proxy Task finishes when task finishes, with task’s completion status being copied to the proxy.
If task fails to complete before the specified timeout period has elapsed, then the proxy Task finishes when the timeout period expires, in Faulted state with a TimeoutException.

In addition to showing how to implement Task.TimeoutAfter, this post will also shed some light on the general thought process that should go into implementing such a feature.

A First Try

Here’s some code that will do the trick:

internal struct VoidTypeStruct { }  // See Footnote #1

static class TaskExtensions
{
    public static Task TimeoutAfter(this Task task, int millisecondsTimeout)
    {
        // tcs.Task will be returned as a proxy to the caller
        TaskCompletionSource<VoidTypeStruct> tcs = 
            new TaskCompletionSource<VoidTypeStruct>();

        // Set up a timer to complete after the specified timeout period
        Timer timer = new Timer(_ => 
        {
            // Fault our proxy Task with a TimeoutException
            tcs.TrySetException(new TimeoutException()); 
        }, null, millisecondsTimeout, Timeout.Infinite);

        // Wire up the logic for what happens when source task completes
        task.ContinueWith(antecedent =>
        {
            timer.Dispose(); // Cancel the timer
            MarshalTaskResults(antecedent, tcs); // Marshal results to proxy
        }, CancellationToken.None, 
            TaskContinuationOptions.ExecuteSynchronously, 
            TaskScheduler.Default);

        return tcs.Task;
    }
}

Simple enough, right? You start a Timer job that faults the proxy Task, and also add a continuation off of the source Task that transfers the completion state of the source to the proxy. The final state of the proxy will therefore depend on which completes first, the Timer job or the source Task.

And by the way, MarshalTaskResults is implemented like this:

internal static void MarshalTaskResults<TResult>(
    Task source, TaskCompletionSource<TResult> proxy)
{
    switch (source.Status)
    {
        case TaskStatus.Faulted:
            proxy.TrySetException(source.Exception);
            break;
        case TaskStatus.Canceled:
            proxy.TrySetCanceled();
            break;
        case TaskStatus.RanToCompletion:
            Task<TResult> castedSource = source as Task<TResult>;
            proxy.TrySetResult(
                castedSource == null ? default(TResult) : // source is a Task
                    castedSource.Result); // source is a Task<TResult>
            break;
    }
}

The “RanToCompletion” handling might seem a little more complicated than it needs to be, but it will allow us to handle Task<TResult> objects correctly (discussed briefly below).

Can We Do Better?

While our first stab at a TimeoutAfter method is functionally correct, we could streamline it and improve its performance. Specifically, notice that our Timer and continuation delegates “capture” variables; this will cause the compiler to allocate special “closure” classes for these delegates behind the scenes, which will slow down our method. To eliminate the need for the closure class allocations, we can pass all “captured” variables in through state variables for those respective calls, like this:

public static Task TimeoutAfter(this Task task, int millisecondsTimeout)
{
    // tcs.Task will be returned as a proxy to the caller
    TaskCompletionSource<VoidTypeStruct> tcs = 
        new TaskCompletionSource<VoidTypeStruct>();

    // Set up a timer to complete after the specified timeout period
    Timer timer = new Timer(state => 
    { 
        // Recover our state data
        var myTcs = (TaskCompletionSource<VoidTypeStruct>)state;

        // Fault our proxy Task with a TimeoutException
        myTcs.TrySetException(new TimeoutException()); 
    }, tcs, millisecondsTimeout, Timeout.Infinite);

    // Wire up the logic for what happens when source task completes
    task.ContinueWith((antecedent,state) =>
    {
        // Recover our state data
        var tuple = 
            (Tuple<Timer, TaskCompletionSource<VoidTypeStruct>>)state;

        // Cancel the timer
        tuple.Item1.Dispose();
        // Marshal results to proxy
        MarshalTaskResults(antecedent, tuple.Item2);    
    }, 
    Tuple.Create(timer,tcs),  // See Footnote #2
    CancellationToken.None,
    TaskContinuationOptions.ExecuteSynchronously,
    TaskScheduler.Default);

    return tcs.Task;
}

Some ad-hoc performance tests show that this little optimization shaves about 12% off of the overhead from the TimeoutAfter method.

What about Edge Cases?

What do we do when the caller specifies a zero timeout, or an infinite timeout? What if the source Task has already completed by the time that we enter the TimeoutAfter method? We can address these edge cases in the TimeoutAfter implementation as follows:

public static Task TimeoutAfter(this Task task, int millisecondsTimeout)
{
    // Short-circuit #1: infinite timeout or task already completed
    if (task.IsCompleted || (millisecondsTimeout == Timeout.Infinite))
    {
        // Either the task has already completed or timeout will never occur.
        // No proxy necessary.
        return task;
    }

    // tcs.Task will be returned as a proxy to the caller
    TaskCompletionSource<VoidTypeStruct> tcs = 
        new TaskCompletionSource<VoidTypeStruct>();

    // Short-circuit #2: zero timeout
    if (millisecondsTimeout == 0)
    {
        // We've already timed out.
        tcs.SetException(new TimeoutException());
        return tcs.Task;
    }

    // Set up a timer to complete after the specified timeout period
    Timer timer = new Timer(state => 
    {
        // Recover your state information
        var myTcs = (TaskCompletionSource<VoidTypeStruct>)state;

        // Fault our proxy with a TimeoutException
        myTcs.TrySetException(new TimeoutException()); 
    }, tcs, millisecondsTimeout, Timeout.Infinite);

    // Wire up the logic for what happens when source task completes
    task.ContinueWith((antecedent, state) =>
    {
        // Recover our state data
        var tuple = 
            (Tuple<Timer, TaskCompletionSource<VoidTypeStruct>>)state;

        // Cancel the Timer
        tuple.Item1.Dispose();

        // Marshal results to proxy
        MarshalTaskResults(antecedent, tuple.Item2);
    }, 
    Tuple.Create(timer, tcs),
    CancellationToken.None,
    TaskContinuationOptions.ExecuteSynchronously,
    TaskScheduler.Default);

    return tcs.Task;
}

Such changes ensure efficient handling of the edge cases associated with TimeoutAfter.

A Different Approach

My colleague Stephen Toub informed me of another potential implementation of Task.TimeoutAfter:

public static async Task TimeoutAfter(this Task task, int millisecondsTimeout)
{
    if (task == await Task.WhenAny(task, Task.Delay(millisecondsTimeout))) 
        await task;
    else
        throw new TimeoutException();
}

The implementation above takes advantage of the new async/await support in .NET 4.5, and is pleasingly concise. However, it does lack some optimizations:

The edge cases described previously are not handled well. (But that could probably be fixed.)
A Task is created via Task.Delay, instead of just a simple timer job.
In the cases where the source Task (task) completes before the timeout expires, no effort is made to cancel the internal timer job that was launched in the Task.Delay call. If the number of “zombie” timer jobs starts becoming significant, performance could suffer.

Nevertheless, it is good to consider the use of async/await support in implementing features like this. Often await will be optimized in ways that simple continuations are not.

What about TimeoutAfter<TResult>?

Suppose that we want to implement the generic version of TimeoutAfter?

public static Task TimeoutAfter<TResult>(
    this Task<TResult> task, 
    int millisecondsTimeout)

It turns out that the implementation of the above would be nearly identical to the non-generic version, except that a TaskCompletionSource<TResult> would be used instead of a TaskCompletionSource<VoidTypeStruct>. The MarshalTaskResults method was already written to correctly handle the marshaling of the results of generic Tasks.

[1] There is no non-generic version of TaskCompletionSource<TResult>. So, if you want a completion source for a Task (as opposed to a Task<TResult>), you still need to provide some throwaway TResult type to TaskCompletionSource. For this example, we’ve created a dummy type (VoidTypeStruct), and we create a TaskCompletionSource<VoidTypeStruct>.

[2] So does it really buy you anything to replace a closure allocation with a tuple allocation? The answer is “yes”. If you were to examine the IL produced from the original code, you would see that both a closure object and a delegate need to be allocated for this call. Eliminating variable capture in the delegate typically allows the compiler to cache the delegate, so in effect two allocations are saved by eliminating variable capture. Thus in this code we’ve traded closure and delegate allocations for a Tuple allocation, so we still come out ahead.

Exception Handling in TPL Dataflow Networks

Cristina Manu — Thu, 10 Nov 2011 00:03:31 GMT

A “dataflow block” is represented by a class implementing the IDataflowBlock interface. The state of a dataflow block is represented by the state of its IDataflowBlock.Completion Task, which itself has a Status property. When a dataflow block is in active state, meaning that it is currently doing processing or may do more processing in the future, its Completion task will be in a non-final state, such as TaskStatus.WaitingForActivation. Once the dataflow block has completed all of its processing, either successfully or due to failure or cancellation, the Completion task’s Status will return one of the three values below:

RanToCompletion
Canceled
Faulted

This blog post will focus on the third state of a completed dataflow block: Faulted.

Current users of the Task Parallel Library (TPL) will notice that the exception model employed by TPL Dataflow follows the same model as TPL. In TPL, if an exception goes unhandled in a Task’s delegate, the Task ends in the Faulted state. Following the same model, in TPL Dataflow if an exception goes unhandled during the processing of a message, the exception will fault the block’s Completion task. The Exception will be available from the block’s Completion Task’s Exception property, and it can be handled as in TPL model. Note, as well, that execution blocks (e.g. ActionBlock) may support parallel processing, such that the block can be processing multiple messages concurrently; this can be configured via the MaxDegreeOfParallelism property on the DataflowBlockOptions used to configure the block when it is constructed. As with parallel loops and PLINQ queries, exceptions in such parallel processing do not forcibly interrupt other concurrent processing, nor is the construct’s processing considered completed the moment the exception occurs; if an exception does occur, the dataflow block will complete only once all processing has quiesced.

Transition to “Faulted” state

If one of the below conditions is met the block will move to Faulted state.

1. Explicit Fault: The invocation of IDataflowBlock.Fault(Exception) will Fault the block. In this case, the exception provided as an argument to the Fault method will be used as the block’s Completion Task’s exception.

BufferBlock<int> block = new BufferBlock<int>();

((IDataflowBlock)block).Fault(new Exception("Test"));

2. Unhandled Delegate Exception: For blocks that run user delegates, e.g. ActionBlock, the user-provided delegate could throw exceptions. Any unhandled exception from these delegates will fault the block.

ActionBlock<int> action = new ActionBlock<int>((x) => 
{ 
    throw new Exception("Test"); 
});

action.Post(1);

try
{
    action.Completion.Wait();
}
catch(AggregateException ex)
{
    Console.WriteLine(ex.InnerException.Message);
}

3. Fault Propagation: The Fault state of a source block can be automatically propagated to linked target blocks if the link between the blocks was configured with a DataflowLinkOptions that had its PropagateCompletion set to true.

TransformBlock<int, int> transformBlock = 
    new TransformBlock<int, int>((x) => 
{ 
    if (x == 1) { throw new Exception("Test"); } else { return x; } 
});

BufferBlock<int> bufferBlock = new BufferBlock<int>();

transformBlock.LinkTo(bufferBlock, 
    new DataflowLinkOptions() { PropagateCompletion= true});

transformBlock.Post(1);
try
{
    bufferBlock.Completion.Wait();
}
catch(AggregateException ex)
{
    Console.WriteLine(ex.InnerException.InnerException.Message);
}

4. Incorrect Interface Implementations: If a block is incorrectly implemented such that one of its interface methods used for inter-block messaging throws an exception, that exception may fault the block with which it’s communicating. For example, a source block might be incorrectly implemented such that its ConsumeMessage method throws. Any linked block that will try to consume the message will become faulted even if the incorrectly implemented block is not faulted. Similar behavior can be observed if an incorrectly implemented target throws from its OfferMessage method.

BufferBlock<int> bufferBlock = new BufferBlock<int>(
    new DataflowBlockOptions() { BoundedCapacity = 1 });

InvalidSource<int> source = new InvalidSource<int>(2);

bufferBlock.Post(1);
//buffer is full so the message will be postponed

source.LinkTo(bufferBlock);

//make room for the new message to be consumed
int firstElement = bufferBlock.Receive();

try
{
    bufferBlock.Completion.Wait();
}
catch(AggregateException ex)
{
    Console.WriteLine(ex.InnerException.Message);
}

public class InvalidSource<T> : ISourceBlock<T>
{
    T message;

    public FaultedSource(T message)
    {
        this.message= message;
    }

    T ISourceBlock<T>.ConsumeMessage(
        DataflowMessageHeader messageHeader, ITargetBlock<T> target, 
        out bool messageConsumed)
    {
        throw new Exception("Test");
    }

    public IDisposable LinkTo(ITargetBlock<T> target, 
        DataflowLinkOptions linkOptions)
    {
        target.OfferMessage(new DataflowMessageHeader(1), 
            this.message, this, true);
        return null;
    }

    void ISourceBlock<T>.ReleaseReservation(
        DataflowMessageHeader messageHeader, ITargetBlock<T> target)
    {
        throw new NotImplementedException();
    }

    bool ISourceBlock<T>.ReserveMessage(
        DataflowMessageHeader messageHeader, ITargetBlock<T> target)
    {
        throw new NotImplementedException();
    }

    public void Complete()
    {
        throw new NotImplementedException();
    }

    public Task Completion
    {
        get { throw new NotImplementedException(); }
    }

    void IDataflowBlock.Fault(Exception exception)
    {
        throw new NotImplementedException();
    }
}

Behaviors of a “Faulted” Block

After a block becomes faulted:

It should stop processing additional messages.
It should clear its message queues, both input and output.
It should fault its Completion task.
If it is a target:

It should decline any further incoming messages.
It should release any reserved messages.
Optionally, it may reserve and then release any postponed messages to notify the sources they will never be needed.

If it is a source:

It should stop offering messages.
It should transfer the exception to any targets currently linked where those links were created with DataflowLinkOptions.PropagateCompletion == true.
It should unlink from currently linked targets

Behaviors of a network with “Faulted” blocks

1. Reserved Messages

In order to avoid message corruption, a faulted block should clear its message queues and move into a Faulted state as soon as possible. There is a single scenario that does not obey to this rule: a source block holding a message reserved by a target. If a block that encounters an internal exception has a message that was reserved by a target, the reserved message must not be dropped, and the block should not be moved into the Faulted state until the message is released or consumed. Here is an example.

BufferBlock<int> bufferBlock = new BufferBlock<int>();
ReservingTarget<int> target = new ReservingTarget<int>();

bufferBlock.LinkTo(target);
bufferBlock.Post(1);
bufferBlock.Post(2);
bufferBlock.Post(3);

//give time to have the message reserved
Thread.Sleep(1000);

((IDataflowBlock)bufferBlock).Fault(new Exception("Test"));
bool completionBlock = bufferBlock.Completion.Wait(2000);
Console.WriteLine("Block is completed {0}. ", completionBlock);

//consume the message and block will be completed
target.ConsumeMessage();

try
{
    bufferBlock.Completion.Wait();
}
catch (AggregateException ex)
{
    Console.WriteLine(ex.InnerException.Message);
}

class ReservingTarget<T> : ITargetBlock<T>
{
    DataflowMessageHeader messageHeader;
    ISourceBlock<T> source;

    DataflowMessageStatus ITargetBlock<T>.OfferMessage(
        DataflowMessageHeader messageHeader, T messageValue, 
        ISourceBlock<T> source, bool consumeToAccept)
    {
        this.messageHeader = messageHeader;
        this.source = source;

        //reserve
        Task.Factory.StartNew(() => 
            { source.ReserveMessage(messageHeader, this); });

        return DataflowMessageStatus.Postponed;
    }

    public void ConsumeMessage()
    {
        bool msgConsumeStatus = false;
        this.source.ConsumeMessage(this.messageHeader, 
            this, out msgConsumeStatus);
    }

    public void Complete()
    {
        throw new NotImplementedException();
    }

    public Task Completion
    {
        get { throw new NotImplementedException(); }
    }

    void IDataflowBlock.Fault(Exception exception)
    {
        throw new NotImplementedException();
    }
}

2. Hanging Networks

Dataflow blocks may be written to expect a specific number of input messages offered to it. Especially in the case where an earlier block in the network might fault, it’s possible such a block may not receive the expected number of messages; in the faulted case, blocks will explicitly drop messages they’ve already processed but haven’t yet forwarded and they’ll decline further messages offered to them. There are two ways to observe that a block in a network faults:

Keep a reference to all the blocks in the network and use Task.WaitAll or Task.WhenAll to wait for them (synchronously or asynchronously). If a block faults, its Completion task will complete in the Faulted state.
Use DataflowLinkOptions with PropagateCompletion == true when building a linear network. That will propagate block completion from source to target. In this case it is enough to wait on the network leaf block.

Behavior of Extensions Methods on Faulted blocks

Choose: If all provided blocks have faulted, the returned task will be completed in the Canceled state.
LinkTo: If the source block is completed (faulted or otherwise), no link should be created. The IDisposable returned from the call will have a Dispose method that’s effectively a nop.
OutputAvailableAsync: If the source block is faulted, the returned Task<Boolean> will be completed in the RanToCompletion state and with its Result equal to false.
Post: If the target block is declining further messages (e.g. because it’s completed, because it experienced an unhandled exception and is shutting down, etc.), the result of the Post method will be false.
Receive: If the source block is faulted, this will throw an InvalidOperationException.
ReceiveAsync: If the source block is faulted, the returned Task will complete in the Faulted state with the InvalidOperationException representing no data available (just as in the Receive case).
SendAsync: As with Post, if the target block is declining further messages (e.g. because it’s completed, because it experienced an unhandeled exception and is shutting down, etc.), the returned task will complete in the RanToCompletion state with its Result equal to false.
TryReceive: Trying to receive from a faulted block will return false.

Exception Handling Design Guidelines for new Dataflow Blocks

It is recommended that new Dataflow blocks are implemented following the below guidelines:

Avoid throwing exceptions in the implementation of the block’s interface methods.
The block should propagate the completion state to a linked target if, and only if, the blocks were linked using DataflowLinkOptions that has its PropagateCompletion property set to true.
Consider faulting the block when an exception is raised during message processing.
Consider cleaning block’s input / output queues when the block moves into the Complete state.
Consider releasing the postponed messages when the block moves into the Complete state.

ConcurrentDictionary Performance Improvements in .NET 4.5

Igor Ostrovsky - MSFT — Tue, 08 Nov 2011 22:16:00 GMT

ConcurrentDictionary is a popular concurrent data structure that was introduced in .NET 4. In the .NET 4.5 release, ConcurrentDictionary gets two performance improvements.

One optimization is related to the way ConcurrentDictionary avoids torn reads and writes. To explain the background, all reference types and some value types are guaranteed to be read and written atomically by the Common Language Runtime, as defined in section 12.6.6 of the ECMA CLI specification. However, large value types such as System.Guid are not read and written atomically. So, if a thread writes values into a System.Guid field while another thread reads the values, the reader thread might observe a partly updated value that consists of bytes from the ‘old’ value mixed with bytes from the ‘new’ value. The reason behind this is that a write or a read of a System.Guid value is not atomic on X86/X64 machines.

To avoid torn reads and writes, the .NET 4 ConcurrentDictionary wraps all values in a node object. To modify a value associated with some key, ConcurrentDictionary allocates a new node object.

In .NET 4.5, ConcurrentDictionary<TKey,TValue> now avoids reallocating nodes on updates if TValue is a reference type or a small primitive value type (Int32, Single, Byte, etc.)

The example below demonstrates the improvement:

static void Example1(int key)
{
    var d = new ConcurrentDictionary<int,int>();
    for(int i = 0; i<10000000; i++)
    {
        d[key] = i;
    }
}

Example1(0) executes about 10% faster in .NET 4.5 than it did in .NET 4.

The difference becomes more prominent when multiple threads are involved:

static void Example2()
{
    Parallel.Invoke(
        () => Example1(0),
        () => Example1(1),
        () => Example1(2),
        () => Example1(3));
}

Example2() calls Example1() on four different threads with different key values. In .NET 4.5, Example2 executes about 30-40% faster than it did .NET 4.

Another ConcurrentDictionary optimization that we introduced in .NET 4.5 is dynamic addition of locks. In .NET 4, the number of locks that protect a ConcurrentDictionary instance is fixed over the lifetime of the instance. By default, the number of locks is based on the number of processors, or alternatively, the user can provide a custom concurrency level.

In practice, a large number of locks is often desirable for maximum throughput. On the other hand, we don’t want to allocate too many lock objects, especially if the ConcurrentDictionary only ends up storing only a small number of items.

In .NET 4.5, ConcurrentDictionary will automatically add more locks as more values are stored in the dictionary. (Note that ConcurrentDictionary will not add more locks if the instance was initialized using one of the constructors that accept a concurrencyLevel argument.)

This example demonstrates the benefit of dynamic lock addition:

static void Example3()
{
    var dict = new ConcurrentDictionary<int, int>();
    Action<int> helper = constant =>
    {
        for (int i = 0; i < 10000000; i++)
        {
            dict[(constant + i) % 100000] = i;
        }
    };

    Parallel.Invoke(
        () => helper(0),
        () => helper(10000),
        () => helper(20000),
        () => helper(30000));
}

This example executes approximately 2.5 – 3X faster in .NET 4.5 compared to .NET 4 and demonstrates the combined effect of both optimizations (cheaper atomic updates and dynamic number of locks) acting together.

Updated Async CTP

Stephen Toub - MSFT — Tue, 01 Nov 2011 15:28:27 GMT

In April, we released the Async CTP Refresh, and since then we've seen fantastic adoption of the technology. We've also seen the technology landscape evolve. Windows Phone 7.5, aka "Mango", was released. Silverlight 5 has had both a Beta and an RC release. And there have been multiple patches to Visual Studio and the .NET Framework. All of these have led to some issues with the Async CTP, which many users have run into and let us know about on the forums.

Some Visual Studio and .NET Framework patches have prevented some of you from installing the Async CTP Refresh, and they've even prevented some of you that already have the Async CTP installed from continuing to use it successfully.
Windows Phone 7.1 SDK enables the development of "background agents". However, these agents are restricted from referencing certain DLLs, some of which the AsyncCtpLibrary_Phone.dll happened to reference. That meant you couldn't use the Async CTP in the development of such agents.
Silverlight 5 now includes Tasks. And while that's awesome, it also means that there's then a conflict between the Task implementation included in Silverlight 5 and the Task implementation included in the CTP's AsyncCtpLibrary_Silverlight.dll, which makes it very challenging to use AsyncCtpLibrary_Silverlight.dll with Silverlight 5.

To address all of these problems, we've just released v3 of the Async CTP. The installer has been tested with all known Visual Studio and .NET Framework patches. The AsyncCtpLibrary_Phone.dll has had the bad references removed so that it may be used to develop background agents. And the CTP includes a new AsyncCtpLibrary_Silverlight5.dll, which may be used with Silverlight 5 RC (AsyncCtpLibrary_Silverlight.dll is still included and may be used for apps using Silverlight 4).

(Especially for those of you that previously installed the Async CTP and that are planning to install the update, Lucian Wischik has some good information on his blog about handling common installation problems.)

Enjoy!

When at last you await

Stephen Toub - MSFT — Tue, 25 Oct 2011 05:36:49 GMT

When you start using async methods heavily, you’ll likely see a particular pattern of composition pop up from time to time. Its structure is typically either of the form:

async Task FooAsync()
{
… // some initialization code without awaits
await BarAsync(…);
}

or of the form:

async Task<T> FooAsync()
{
… // some initialization code without awaits
return await BarAsync(…);
}

A concrete example of this might be a FlushAsync method on a custom buffered stream, where the stream maintains a byte[] of data to write out, along with the number of bytes buffered, and flushing the buffer involves resetting the buffered count and writing the data out to the underlying wrapped stream, e.g.

public override async Task FlushAsync()
{
    int numBuffered = m_numBuffered;
    m_numBuffered = 0;
    await m_wrappedStream.WriteAsync(m_buffer, 0, numBuffered);
}

When we find code that follows this particular pattern, we can employ an alternative. Instead of the above, we could write this FlushAsync method as follows:

public override Task FlushAsync()
{
    int numBuffered = m_numBuffered;
    m_numBuffered = 0;
    return m_wrappedStream.WriteAsync(m_buffer, 0, numBuffered);
}

The key difference here is that this method is no longer using the compiler support for writing async methods. By removing the “async” keyword, this is just an ordinary method, rather than one that the compiler rewrites into a state machine. And along with that, we’re now returning a Task created by the body of the method rather than relying on the compiler to create one for us.

Why would we want to do this? Purely as an optimization. If the particular code in question is very performance-sensitive, you can save some cycles by avoiding the overhead necessary to support compiler-generated async methods. When you use the async/await keywords, the compiler and Framework collude to generate a Task on your behalf to represent all of the async method’s processing. In this case, however, we’re being handed a task to represent the last statement in the method, and thus it’s in effect already a representation of the entire method’s processing (as long as nothing in the preamble could throw an exception, since such an exception would directly escape the method rather than being stored into the returned task). Since it already provides the representation we need, we can return it untouched to our caller, saving a level of indirection.

For more information on performance and async methods, see the October 2011 issue of MSDN Magazine.

Task.Run vs Task.Factory.StartNew

Stephen Toub - MSFT — Mon, 24 Oct 2011 20:33:11 GMT

In .NET 4, Task.Factory.StartNew was the primary method for scheduling a new task. Many overloads provided for a highly configurable mechanism, enabling setting options, passing in arbitrary state, enabling cancellation, and even controlling scheduling behaviors. The flip side of all of this power is complexity. You need to know when to use which overload, what scheduler to provide, and the like. And “Task.Factory.StartNew” doesn’t exactly roll off the tongue, at least not quickly enough for something that’s used in such primary scenarios as easily offloading work to background processing threads.

So, in the .NET Framework 4.5 Developer Preview, we’ve introduced the new Task.Run method. This in no way obsoletes Task.Factory.StartNew, but rather should simply be thought of as a quick way to use Task.Factory.StartNew without needing to specify a bunch of parameters. It’s a shortcut. In fact, Task.Run is actually implemented in terms of the same logic used for Task.Factory.StartNew, just passing in some default parameters. When you pass an Action to Task.Run:

Task.Run(someAction);

that’s exactly equivalent to:

Task.Factory.StartNew(someAction,
CancellationToken.None, TaskCreationOptions.DenyChildAttach, TaskScheduler.Default);

In this way, Task.Run can and should be used for the most common cases of simply offloading some work to be processed on the ThreadPool (what TaskScheduler.Default targets). That doesn’t mean Task.Factory.StartNew will never again be used; far from it. Task.Factory.StartNew still has many important (albeit more advanced) uses. You get to control TaskCreationOptions for how the task behaves. You get to control the scheduler for where the task should be queued to and run. You get to use overloads that accept object state, which for performance-sensitive code paths can be used to avoid closures and the corresponding allocations. For the simple cases, though, Task.Run is your friend.

Task.Run provides eight overloads, to support all combinations of the following:

Task vs Task<TResult>
Cancelable vs non-cancelable
Synchronous vs asynchronous delegate

The first two bullets should be self-explanatory. For the first bullet, there are overloads that return Task (for operations that don’t have a result) and there are overloads that return Task<TResult> (for operations that have a result of type TResult). There are also overloads that accept a CancellationToken, which enables the Task Parallel Library (TPL) to transition the task to a Canceled state if cancellation is requested prior to the task beginning its execution.

The third bullet is more interesting, and is directly related to the async language support in C# and Visual Basic in Visual Studio 11. Let’s consider Task.Factory.StartNew for a moment, as that will help to highlight what this distinction is. If I write the following call:

var t = Task.Factory.StartNew(() =>
{
Task inner =Task.Factory.StartNew(() => {});
return inner;
});

the type of ‘t’ is going to be Task<Task>; the task’s delegate is of type Func<TResult>, TResult in this case is a Task, and thus StartNew is returning a Task<Task>. Similarly, if I were to change that to be:

var t = Task.Factory.StartNew(() =>
{
Task<int> inner = Task.Factory.StartNew(() => 42));
return inner;
});

the type of ‘t’ is now going to be Task<Task<int>>. The task’s delegate is Func<TResult>, TResult is now Task<int>, and thus StartNew is returning Task<Task<int>>. Why is this relevant? Consider now what happens if I write the following:

var t = Task.Factory.StartNew(async delegate
{
await Task.Delay(1000);
return 42;
});

By using the async keyword here, the compiler is going to map this delegate to be a Func<Task<int>>: invoking the delegate will return the Task<int> to represent the eventual completion of this call. And since the delegate is Func<Task<int>>, TResult is Task<int>, and thus the type of ‘t’ is going to be Task<Task<int>>, not Task<int>.

To handle these kinds of cases, in .NET 4 we introduced the Unwrap method. Unwrap has two overloads, which are both extensions methods, one on type Task<Task> and one on type Task<Task<TResult>>. We called this method Unwrap because it, in effect, “unwraps” the inner task that’s returned as the result of the outer task. Calling Unwrap on a Task<Task> gives you back a new Task (which we often refer to as a proxy) which represents the eventual completion of the inner task. Similarly, calling Unwrap on a Task<Task<TResult>> gives you back a new Task<TResult> which represents the eventual completion of that inner task. (In both cases, if the outer task is Faulted or Canceled, there is no inner task, since there’s no result from a task that doesn’t run to completion, so the proxy task then represents the state of the outer task.) Going back to the prior example, if I wanted ‘t’ to represent the return value of that inner task (in this case, the value 42), I could write:

var t = Task.Factory.StartNew(async delegate
{
await Task.Delay(1000);
return 42;
}).Unwrap();

The ‘t’ variable will now be of type Task<int>, representing the result of that asynchronous invocation.

Enter Task.Run. Because we expect it to be so common for folks to want to offload work to the ThreadPool, and for that work to use async/await, we decided to build this unwrapping functionality into Task.Run. That’s what’s referred to by the third bullet above. There are overloads of Task.Run that accept Action (for void-returning work), Func<TResult> (for TResult-returning work), Func<Task> (for void-returning async work), and Func<Task<TResult>> (for TResult-returning async work). Internally, then, Task.Run does the same kind of unwrapping that’s shown with Task.Factory.StartNew above. So, when I write:

var t = Task.Run(async delegate
{
await Task.Delay(1000);
return 42;
});

the type of ‘t’ is Task<int>, and the implementation of this overload of Task.Run is basically equivalent to:

var t = Task.Factory.StartNew(async delegate
{
await Task.Delay(1000);
return 42;
}, CancellationToken.None, TaskCreationOptions.DenyChildAttach, TaskScheduler.Default).Unwrap();

As mentioned before, it’s a shortcut.

All of this then means that you can use Task.Run either with either regular lambdas/anonymous methods or with async lambdas/anonymous methods, and the right thing will just happen. If I wanted to offload this work to the ThreadPool and await its result, e.g.

int result = await Task.Run(async () =>
{
await Task.Delay(1000);
return 42;
});

the type of result will be int, just as you’d expect, and approximately one second after this work is invoked, the result variable be set to the value 42.

Interestingly, the new await keyword can almost be thought of as a language equivalent to the Unwrap method. So, if we return back to our Task.Factory.StartNew example, I could rewrite the last snippet above as follows using Unwrap:

int result = await Task.Factory.StartNew(async delegate
{
await Task.Delay(1000);
return 42;
}, CancellationToken.None, TaskCreationOptions.DenyChildAttach, TaskScheduler.Default).Unwrap();

or, instead of using Unwrap, I could use a second await:

int result = await await Task.Factory.StartNew(async delegate
{
await Task.Delay(1000);
return 42;
}, CancellationToken.None, TaskCreationOptions.DenyChildAttach, TaskScheduler.Default);

“await await” here is not a typo. Task.Factory.StartNew is returning a Task<Task<int>>. Await’ing that Task<Task<int>> returns a Task<int>, and awaiting that Task<int> returns an int. Fun, right?

New articles on async/await in MSDN Magazine

Stephen Toub - MSFT — Mon, 03 Oct 2011 15:47:41 GMT

The October 2011 issue of MSDN Magazine is now available online. In it, you can find three articles about the new async/await features of C# and Visual Basic. While the articles can stand alone, they were written with each other in mind in order to provide a 1-2-3 on-ramp into the world of asynchronous programming with these new language and library features.

Here they are. If you read them all, we recommend you read them in the following order:

Asynchronous Programming: Easier Asynchronous Programming with the New Visual Studio Async CTP by Eric Lippert
Asynchronous Programming: Pause and Play with Await by Mads Torgersen
Asynchronous Programming: Understanding the Costs of Async and Await by Stephen Toub

Enjoy!