Rather than finishing this series on asynchrony before my Monospace 2011 talk on the same subject, I got busy finishing the materials for the talk instead. The slides and example code from the talk can be found here, and the video will be posted on InfoQ in the near future. Since the examples are slightly different and the slides less explanatory, I will continue this series of posts as a more in-depth version of the content in the slides.

In the last post, Exit… Screen Right, I talked about using the CPS (continuation passing style) to chain asynchronous methods together. While this is a proven way of dealing with asynchrony, it puts the how in front of the what, which makes our code harder to maintain and debug. Ideally we'd continue writing our code in the regular sequential style we've always used but steps that do not happen synchronously, suspend the flow and resume once the step is completed instead of blocking their thread. With a regular method that's not possible, since it controls the flow of execution until it exits, but there exists a variation of the trusty method (which really is just a subroutine), called coroutine:

Coroutines are computer program components that generalize subroutines to allow multiple entry points for suspending and resuming execution at certain locations

– Wikipedia

With Coroutines, we can suspend the current execution flow, allow an async action to complete and then have its continuation resumes our flow of execution. Great. Except .NET doesn't have Coroutines, right? Not exactly, but it does have generators for building Iterators, in the form of IEnumerator<T> and the yield keyword. Since Generators are a specialized form of Coroutines it is possible to use them to mimic Coroutine behavior. In addtion, the AsyncCTP for C# 5.0 introduces the async/await constructs, which provide a better syntax for suspending and resuming a method.

DReAM Coroutines are built on top an inversion of the Iterator pattern, which allowed us to introduce them in .NET 2.0, while Task based Coroutines won't be possible in production until .NET 5.0 ships. Both provide us the ability to write sequential code that suspends on asynchronous work, preserving our workflow.

Review of our Async Workflow Methods

Below are the asynchronous method signatures from last time:

1. GetUserRecentArticles

This method implements the asynchronous workflow.

// Result
Result<DreamMessage> GetUserRecentArticles(int userId, int limit, Result<DreamMessage> response)
// Task
Task<DreamMessage> GetUserRecentArticles(int userId, int limit)

DreamMessage is DReAM's symmetric Http Request and Response data container, and serves as the value of the synchronization handle seen by the outside caller. The Http Server would call GetUserRecentArticles to start the asynchronous chain and use the synchronization handle to retrieve the final result.

2. GetFeedsForUserFromDb

This method uses asynchronous database queries to get the list of Uri's the user is subscribed to:

// Result
Result<IEnumerable<XUri>> GetFeedsForUserFromDb(int userId, Result<IEnumerable<XUri>> result)
// Task
Task<IEnumerable<XUri>> GetFeedsForUserFromDb(int userId)

3. FetchFeed

This method uses an asynchronous web request to fetch the feed for a given Uri. We will be multiplexing this call to fetch multiple feeds at once.

// Result
Result<XDoc> FetchFeed(XUri uri, Result<XDoc> result)
//Task
Task<XDoc> FetchFeed(XUri uri)

4. FilterViaDb

This method is called once for each feed that we've fetched to turn into a list of only the articles not yet seen by the user

// Result
Result<IEnumerable<XDoc>> FilterViaDb(XDoc doc, Result<IEnumerable<XDoc>> result)
// Task
Task<IEnumerable<XDoc>> FilterViaDb(XDoc doc)

5. MarkAsRead

Step 5 combines the articles, applies the limit and marks the articles to be returned as read. Only the last part is an asynchronous operation, the rest can be performed inline. The method MarkAsRead makes asynchronous calls to the database to insert the canonical Uri's of the articles to be returned to mark them as read.

//Result
Result MarkAsRead(int userId, IEnumerable<XDoc> aggregatedArticles, Result result)
// Task
Task MarkAsRead(int userId, IEnumerable<XDoc> aggregatedArticles)

DReAM coroutine

A DReAM Coroutine must produce an Enumerator of type IYield, which is the coordination construct for Coroutines and is implemented by Result, i.e. any method returning a Result can be yielded to in a Coroutine. Since this signature requirement would change the expected signature of GetUserRecentArticles from the DReAM Asynchronous Method Pattern and the usage of a Coroutine is really an implementation detail, we hide the actual Coroutine:

private Result<DreamMessage> GetUserRecentArticles(int userId, int limit, Result<DreamMessage> response) {
  return Coroutine.Invoke(GetUserRecentArticles_Co, userId, limit, response);
}

We use Coroutine.Invoke to execute the enumerator that GetUserRecentArticles_Co returns. The actual implementation now has the familiar sequential flow we are used to:

public Yield GetUserRecentArticles_Co(int userId, int limit, Result response) {

    // yield execution to fetch the user's feed uris from the DB
    IEnumerable uris = null;
    yield return GetFeedsForUserFromDb(userId, new Result>()).Set(x => uris = x);

    // start to fetch all feeds
    var feeds = uris.Select(uri => FetchFeed(uri, new Result())).ToList();

    // yield execution until all fetch calls have signaled completion
    yield return feeds.Join(new Result());

    // start to filter all feeds to only unread
    var allUnreadArticles = feeds
        .Select(feedResult => FilterViaDb(feedResult.Value, new Result>()))
        .ToList();

    // yield execution until all filter calls have signaled completion
    yield return allUnreadArticles.Join(new Result());

    // combine all filtered articles
    var combinedArticles = (from unreadResult in allUnreadArticles
                            let unreadArticles = unreadResult.Value
                            from article in unreadArticles
                            select article).Take(limit).ToList();

    // yield execution to async method marking all remaining articles as read
    yield return MarkAsRead(userId, combinedArticles, new Result());
    var aggregatedDoc = new XDoc("articles").AddAll(combinedArticles);

    // signal response with aggregated article document
    response.Return(DreamMessage.Ok(aggregatedDoc));
}

Instead of chaining .WhenDone on each async method, we now call them with yield return instead, which resumes the body of the method after the async call completes. We still use response.Return to signal completion, since the Iterator doesn't regular return value. The one caveat causing some syntactic cruft is that we cannot create and assign the value in the yield return. Instead we have to declare it before and use the .Set() extension method, which unwraps the result, allowing us to capture it with a lambda.

Coroutine flow with async/await

async/await allows the same coroutine flow we create with an Iterator and Result, but since it is custom built into the compiler around Task, it achieves additional syntactic magic for very readable and natural code:

public async Task GetUserRecentArticles(int userId, int limit) {

    // Await the fetching of the user's feed uris from the DB
    var uris = await GetFeedsForUserFromDb(userId);

    // start fetch all feeds
    var feedTasks = uris.Select(FetchFeed).ToList();

    // Await all feeds being fetched
    var feeds = await  TaskEx.WhenAll(feedTasks);

    // start to filter all feeds to only unread
    var unreadArticleTasks = feeds.Select(FilterViaDb).ToList();

    // Await feeds completing filtering
    var allUnreadArticles = await TaskEx.WhenAll(unreadArticleTasks);

    // combine all filtered articles
    var combinedArticles = (from unreadArticles in allUnreadArticles
                            from article in unreadArticles
                            select article).Take(limit).ToList();

    // Await the database call to mark all remaining articles as read
    await MarkAsRead(userId, combinedArticles);
    var aggregatedDoc = new XDoc("articles").AddAll(combinedArticles);

    // return the final result, which in turn sets it on the Task
    // that had already been returned for us
    return DreamMessage.Ok(aggregatedDoc);
}

With async/await, the only difference between a workflow where all of our calls are synchronous is that we have an await in front of every async method now. That's a really low syntax tax to pay for completely non-blocking workflows! The other thing you may notice is that the signature of the method is Task<DreamMessage> while we return just a DreamMessage. This is possible, because the method is marked with async, which tells the compiler to rewrite the method body as a series of continuations interrupted by await barriers.

But how does it work?

Both yield and async/await are constructs that tell the compiler that the body of the subroutine should be rewritten as a state machine. They make it easy for us to think of the flow of execution as a regular sequential flow, while handling the complexities of tracking the local scope when execution is suspended and resumed. In simplistic terms, a new anonymous class is generated with members to capture the locally available variables of the method. The body of the method is broken up into linear blocks of execution at the yield/await barriers. This way each block can be run until the next barrier, store to which point it was last run to and exit and upon the next invocation continue running the next block.

In my next post, I will break down how you can use this behavior to roll your own Coroutines instead of just using yield to return values in a sequence or await to wait for an asynchronous method to complete.