Async C# 5: Using Async Pattern with WCF and Silverlight

Sometime ago I published a way to use WCF service with the new Async Pattern in C#. The problem is that this approach won’t work with Silverlight. More specifically, TaskFactory is not available in Silverlight.

So.. here is one solution (it does not provide full functionality as in TaskFactory, but you can manage 90% of use cases):

MyTaskFactory

public static class MyTaskFactory
{
    public static Task<TResult> FromAsync<TResult>(
        IAsyncResult asyncResult, Func<IAsyncResult, TResult> endMethod)
    {
        var completionSource = new TaskCompletionSource<TResult>();
        if (asyncResult.IsCompleted)
        {
            completionSource.TrySetResult(asyncResult, endMethod);
        }
        else
        {
            System.Threading.ThreadPool.RegisterWaitForSingleObject(
                asyncResult.AsyncWaitHandle,
                (state, timeOut) =>
                {
                    completionSource.TrySetResult(asyncResult, endMethod);
                }, null, -1, true);
        }

        return completionSource.Task;
    }

    static void TrySetResult<TResult>(
        this TaskCompletionSource<TResult> completionSource,
        IAsyncResult asyncResult, Func<IAsyncResult, TResult> endMethod)
    {
        try
        {
            var result = endMethod(asyncResult);
            completionSource.TrySetResult(result);
        }
        catch (OperationCanceledException)
        {
            completionSource.TrySetCanceled();
        }
        catch (Exception genericException)
        {
            completionSource.TrySetException(genericException);
        }
    }
}

The previous code fragment defines a helper method that behaves similar to TaskFactory.

Usage example:

Usage:

var client = new MyServiceClient();
// EchoTaskAsync is an extension method
var result = await client.EchoTaskAsync("zzzz");

And here is the service extension for MyServiceClient (Note that I use the service interface contract instead of the ClientBase<> as parameter since we need access to Begin/End methods):

Service Async Extensions

public static class ServiceExtensions
{
    public static Task<string> EchoTaskAsync(
        this IMyService client, string text)
    {
        return MyTaskFactory.FromAsync<string>(
            client.BeginEcho(text, null, null),
            ar => client.EndEcho(ar));
    }
}

Try it and let me know if it works for you..

Asynchrony in C# 5: Dataflow Async Logger Sample

Check out this (very simple) code examples for TPL Dataflow.

Suppose you are developing an Async Logger to register application events to different sinks or log writers.

The logger architecture would be as follow:

Note how blocks can be composed to achieved desired behavior. The BufferBlock<T> is the pool of log entries to be process whereas linked ActionBlock<TInput> represent the log writers or sinks.

The previous composition would allows only one ActionBlock to consume entries at a time.

Implementation code would be something similar to (add reference to System.Threading.Tasks.Dataflow.dll in %User Documents%\Microsoft Visual Studio Async CTP\Documentation):

TPL Dataflow Logger

var bufferBlock = new BufferBlock<Tuple<LogLevel, string>>();

ActionBlock<Tuple<LogLevel, string>> infoLogger =
    new ActionBlock<Tuple<LogLevel, string>>(
        e => Console.WriteLine("Info: {0}", e.Item2));

ActionBlock<Tuple<LogLevel, string>> errorLogger =
    new ActionBlock<Tuple<LogLevel, string>>(
        e => Console.WriteLine("Error: {0}", e.Item2));

bufferBlock.LinkTo(infoLogger, e => (e.Item1 & LogLevel.Info) != LogLevel.None);
bufferBlock.LinkTo(errorLogger, e => (e.Item1 & LogLevel.Error) != LogLevel.None);

bufferBlock.Post(new Tuple<LogLevel, string>(LogLevel.Info, "info message"));
bufferBlock.Post(new Tuple<LogLevel, string>(LogLevel.Error, "error message"));

Note the filter applied to each link (in this case, the Logging Level selects the writer used). We can specify message filters using Predicate functions on each link.

Now, the previous sample is useless for a Logger since Logging Level is not exclusive (thus, several writers could be used to process a single message).

Let´s use a Broadcast<T> buffer instead of a BufferBlock<T>.

Broadcast Logger

var bufferBlock = new BroadcastBlock<Tuple<LogLevel, string>>(
    e => new Tuple<LogLevel, string>(e.Item1, e.Item2));

ActionBlock<Tuple<LogLevel, string>> infoLogger =
    new ActionBlock<Tuple<LogLevel, string>>(
        e => Console.WriteLine("Info: {0}", e.Item2));

ActionBlock<Tuple<LogLevel, string>> errorLogger =
    new ActionBlock<Tuple<LogLevel, string>>(
        e => Console.WriteLine("Error: {0}", e.Item2));

ActionBlock<Tuple<LogLevel, string>> allLogger =
    new ActionBlock<Tuple<LogLevel, string>>(
    e => Console.WriteLine("All: {0}", e.Item2));

bufferBlock.LinkTo(infoLogger, e => (e.Item1 & LogLevel.Info) != LogLevel.None);
bufferBlock.LinkTo(errorLogger, e => (e.Item1 & LogLevel.Error) != LogLevel.None);
bufferBlock.LinkTo(allLogger, e => (e.Item1 & LogLevel.All) != LogLevel.None);

bufferBlock.Post(new Tuple<LogLevel, string>(LogLevel.Info, "info message"));
bufferBlock.Post(new Tuple<LogLevel, string>(LogLevel.Error, "error message"));

As this block copies the message to all its outputs, we need to define the copy function in the block constructor. In this case we create a new Tuple, but you can always use the Identity function if passing the same reference to every output.

Try both scenarios and compare the results.

Asynchrony in C# 5 (Part II)

This article is a continuation of the series of asynchronous features included in the new Async CTP preview for next versions of C# and VB. Check out Part I for more information.

So, let’s continue with TPL Dataflow:

1. Asynchronous functions
2. TPL Dataflow
3. Task based asynchronous Pattern

Part II: TPL Dataflow

Definition (by quote of Async CTP doc): “TPL Dataflow (TDF) is a new .NET library for building concurrent applications. It promotes actor/agent-oriented designs through primitives for in-process message passing, dataflow, and pipelining. TDF builds upon the APIs and scheduling infrastructure provided by the Task Parallel Library (TPL) in .NET 4, and integrates with the language support for asynchrony provided by C#, Visual Basic, and F#.”

This means: data manipulation processed asynchronously.

“TPL Dataflow is focused on providing building blocks for message passing and parallelizing CPU- and I/O-intensive applications”.

Data manipulation is another hot area when designing asynchronous and parallel applications: how do you sync data access in a parallel environment? how do you avoid concurrency issues? how do you notify when data is available? how do you control how much data is waiting to be consumed? etc. 

Dataflow Blocks

TDF provides data and action processing blocks. Imagine having preconfigured data processing pipelines to choose from, depending on the type of behavior you want. The most basic block is the BufferBlock<T>, which provides an storage for some kind of data (instances of <T>).

So, let’s review data processing blocks available. Blocks a categorized into three groups:

1. Buffering Blocks
2. Executor Blocks
3. Joining Blocks

Think of them as electronic circuitry components :)..

1. BufferBlock<T>: it is a FIFO (First in First Out) queue. You can Post data to it and then Receive it synchronously or asynchronously. It synchronizes data consumption for only one receiver at a time (you can have many receivers but only one will actually process it).

2. BroadcastBlock<T>: same FIFO queue for messages (instances of <T>) but link the receiving event to all consumers (it makes the data available for consumption to N number of consumers). The developer can provide a function to make a copy of the data if necessary.

3. WriteOnceBlock<T>: it stores only one value and once it’s been set, it can never be replaced or overwritten again (immutable after being set). As with BroadcastBlock<T>, all consumers can obtain a copy of the value.

4. ActionBlock<TInput>: this executor block allows us to define an operation to be executed when posting data to the queue. Thus, we must pass in a delegate/lambda when creating the block. Posting data will result in an execution of the delegate for each data in the queue.

You could also specify how many parallel executions to allow (degree of parallelism).

5. TransformBlock<TInput, TOutput>: this is an executor block designed to transform each input, that is way it defines an output parameter. It ensures messages are processed and delivered in order.

6. TransformManyBlock<TInput, TOutput>: similar to TransformBlock but produces one or more outputs from each input.

7. BatchBlock<T>: combines N single items into one batch item (it buffers and batches inputs).

8. JoinBlock<T1, T2, …>: it generates tuples from all inputs (it aggregates inputs). Inputs could be of any type you want (T1, T2, etc.).

9. BatchJoinBlock<T1, T2, …>: aggregates tuples of collections. It generates collections for each type of input and then creates a tuple to contain each collection (Tuple<IList<T1>, IList<T2>>).

Next time I will show some examples of usage for each TDF block.

* Images taken from Microsoft’s Async CTP documentation.

Async CTP (C# 5): How to make WCF work with Async CTP

If you have recently downloaded the new Async CTP you will notice that WCF uses Async Pattern and Event based Async Pattern in order to expose asynchronous operations.

In order to make your service compatible with the new Async/Await Pattern try using an extension method similar to the following:

WCF Async/Await Method

public static class ServiceExtensions
{
    public static Task<DateTime> GetDateTimeTaskAsync(this Service1Client client)
    {
        return Task.Factory.FromAsync<DateTime>(
            client.BeginGetDateTime(null, null),
            ar => client.EndGetDateTime(ar));
    }
}

The previous code snippet adds an extension method to the GetDateTime method of the Service1Client WCF proxy.

Then used it like this (remember to add the extension method’s namespace into scope in order to use it):

Code Snippet

var client = new Service1Client();
var dt = await client.GetDateTimeTaskAsync();

Replace the proxy’s type and operation name for the one you want to await.

Asynchrony in C# 5 (Part I)

I’ve been playing around with the new Async CTP preview available for download from Microsoft. It’s amazing how language trends are influencing the evolution of Microsoft’s developing platform. Much effort is being done at language level today than previous versions of .NET.

In these post series I’ll review some major features contained in this release:

1. Asynchronous functions
2. TPL Dataflow
3. Task based asynchronous Pattern

Part I: Asynchronous Functions

This is a mean of expressing asynchronous operations. This kind of functions must return void or Task/Task<> (functions returning void let us implement Fire & Forget asynchronous operations). The two new keywords introduced are async and await.

• async: marks a function as asynchronous, indicating that some part of its execution may take place some time later (after the method call has returned). Thus, all async functions must include some kind of asynchronous operations. This keyword on its own does not make a function asynchronous thought, its nature depends on its implementation.
• await: allows us to define operations inside a function that will be awaited for continuation (more on this later).

Async function sample:

Async/Await Sample

async void ShowDateTimeAsync()
{
    while (true)
    {
        var client = new ServiceReference1.Service1Client();
        var dt = await client.GetDateTimeTaskAsync();
        Console.WriteLine("Current DateTime is: {0}", dt);
        await TaskEx.Delay(1000);
    }
}

The previous sample is a typical usage scenario for these new features. Suppose we query some external Web Service to get data (in this case the current DateTime) and we do so at regular intervals in order to refresh user’s UI. Note the async and await functions working together.

The ShowDateTimeAsync method indicate its asynchronous nature to the caller using the keyword async (that it may complete after returning control to its caller). The await keyword indicates the flow control of the method will continue executing asynchronously after client.GetDateTimeTaskAsync returns. The latter is the most important thing to understand about the behavior of this method and how this actually works.

The flow control of the method will be reconstructed after any asynchronous operation completes (specified with the keyword await). This reconstruction of flow control is the real magic behind the scene and it is done by C#/VB compilers. Note how we didn’t use any of the regular existing async patterns and we’ve defined the method very much like a synchronous one.

Now, compare the following code snippet  in contrast to the previuous async/await:

Traditional UI Async

void ComplicatedShowDateTime()
{
    var client = new ServiceReference1.Service1Client();
    client.GetDateTimeCompleted += (s, e) =>
    {
        Console.WriteLine("Current DateTime is: {0}", e.Result);
        client.GetDateTimeAsync();
    };
    client.GetDateTimeAsync();
}

The previous implementation is somehow similar to the first shown, but more complicated. Note how the while loop is implemented as a chained callback to the same method (client.GetDateTimeAsync) inside the event handler (please, do not do this in your own application, this is just an example). 

How it works? Using an state workflow (or jump table actually), the compiler expands our code and create the necessary steps to execute it, resuming pending operations after any asynchronous one.

The intention of the new Async/Await pattern is to let us think and code as we normally do when designing and algorithm. It also allows us to preserve the logical flow control of the program (without using any tricky coding patterns to accomplish this). The compiler will then create the necessary workflow to execute operations as the happen in time.

Webcast MSDN: El Futuro de C# y Visual Basic

I will be presenting an screencast tomorrow about next versions of C# and VB.NET. We will review some exciting new features.

Register here.