11 Jan 12

Rolling your own coroutines

This post marks the conclusion of the async programming series I originally started as lead in for my monospace 2011 talk. The remaining subject, as promised in my last article, is the creation of tradtional coroutines using .NET iterators. The example code and Monospace slides talking about it can be found on github. I had hoped to time it with a definite date for async/await arriving in .NET but WinRT has kind of derailed future releases, it would seem.

But before I get into, let's roll up the posts that got us here:

The canonical example of coroutines are producer/consumer pipelines, in which each stage of the pipeline does some amount of work and yields execution to the next stage once it cannot continue. The benefits of this type of processing pipeline is avoidance of locks and maximum utilization of the processing power given to it, in addition to the loose coupling of the pipeline that easily let you add or remove steps in the chain. 

For this processing to work the coroutines must all have the same "shape", e.g. signature. For the Iterator based coroutines I will cover first, I chose the following shape:

The coroutine accepts the Coordinator and yields once it cannot continue with any more work. The Coordinator contains the shared state T that all coroutines have access to. Since only one coroutine is ever executed at a time, the state can be mutated without locks.

The example we'll use is a producer of a 3x4 matrix and a consumer that outputs that matrix transpoed to 6x2. To show the ability to add random stages to this chain, we will have another coroutine that will take every value and square it.

We create a Coordinator<int[]>, i.e. our state is a block of ints, requiring a coroutine signature of IEnumerator Coroutine<T>(Coordinator<T> coordinator), but since the Coordinator accepts the coroutines as an array of Funcs, we can curry signatures  of different shapes into the required shape. In our example we capture a source and destination for the Producer and Consumer coroutines, respectively, while the Exponetiator already has the required shape. 

Let's look at the Producer:

The Producer takes the input matrix and writes it one row at a time into the state object. After each row, the producer yields execution to the next coroutine. We yield null, since the iterator is just used for its side-effect of letting us suspend a method mid-execution and continue later. And that's the magic of it. Yes, it looks like a regular method, but it really does exit and re-enter multiple times. Every time yield is called, the state of the method is suspended while another coroutine gets to run, and once resumed, the method continues on with all its local state from the point right after the yield. We are stopping in the middle of a loop, letting someone else go and then continue on from that exact same place in the loop, all without blocking a thread.

The Consumer is provided the output matrix and starts a loop that will continue until its output matrix is filled, reading the block of integers from the state, writing it to the new matrix and yielding its execution to the coordinator, so that the a fresh set of integers can be provided.

The last coroutine is the Exponentiator which just continues to square every value in the state integer block each time it is resumed and then yields execution until provided a fresh set of integers to square.

Finally, Coordinator<T> allows the construction of these arbitrary co-operative processing chains. It is simply an inversion of a regular iterator. Instead of receiving a new value, one at a time from an enumerator, we have one enumerator per coroutine that we can command to run to its next yield point with MoveNext(). MoveNext() will return  false if it was a yield break, signaling that the coroutine is done or true if it was a yield return, signaling that the coroutine is willing to continue in which case we put the enumerator into our queue, pick the next coroutine from the queue and compel it resume.

In the  github project I also implemented the same coroutines using async/await. The one benefit that implementation has over Iterator based coroutines is that once inside an async method, you can await any other async method. Allowing these types of coroutines to suspend the execution chain to let some other async task (such as a web request) execute.

While interesting and useful in a limited set of circumstances the coroutines introduced in this article are not likely going to find themselves into your next project. Far more useful and applicable in any application liable to block on I/O are the asynchronous workflows introduced in Asynchrony and Sequential Workflows since these types of coroutines allow us to use traditional sequential coding styles while never blocking on asynchronous operations.

19 Aug 11

Asynchrony and sequential workflows

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.

27 Jul 11

Watched Pages have been merged with Subscriptions

You might have noticed that the Watched Pages link in your “More” menu. As I mentioned in a previous post, that’s intentional. There has been confusion between page subscriptions (where you can subscribe for notifications to page/site changes) and Watched Pages (where you can see what pages you’re watching and subscribe to them via RSS, for example) for some time now. Obviously, this was causing great confusion among new users so we decided to merge the functionality of the two services. Now, instead of having a watched pages page, you just have a Subscriptions page located here: http://developer.mindtouch.com/Special:Subscriptions

As a result, the watched pages you were watching have automatically been converted to Subscribed pages. You can manage subscriptions more easily now as well. Learn more here: http://help.mindtouch.us/kb/Managing_Notifications

Note: the Subscriptions link is missing from the Tools menu on Developer.MindTouch.com. This is a known error in the skin being used here. We will likely push a fix live today. In the meantime, you can directly navigate to the subscriptions page by using the link above.

29 Jun 11

New 10.1 Feature: Advanced Search

UPDATE:  Please accept our sincerest apologies for any miscommunication, though the new Advanced Search feature is currently only available within our TCS Cloud offering.  We are looking to include this functionality in a future release of MindTouch Platform v 10.1.

MindTouch Core, Platform and TCS all ship with a new advanced search feature that takes a normal query and gives you a dedicated UI for going into further detail about what you want. Advanced search allows you to refine your queries by including “all of these words” “exact phrase” “any of these words” or “been tagged with.” You can also filter results by “do not have these words” “by author” “by type:pages, images, comments, or documents.” If you are a TCS customer, these results are going to feed into your search analytics as well.

You can read more about advanced search and see some screenshots over on help.mindtouch.us.

Tags:
29 Jun 11

New 10.1 Feature: Spell-Check

We’re not all great spellers. Let’s face it, most of us are pretty bad spellers. We’re also not great at grammar. When you ask for collaboration amongst your community, you have to take the bad with the good – spelling and grammar included. Wouldn’t it be nice if your collaboration platform had a spell-check and grammar-check built into the editor?

In the 10.1 release of Core, Platform and TCS, MindTouch now ships with that capability out of the box. Look for the ABC button at the top of your editor window and click it. Wait a sec while the spell-check reads your content and then start to click on the items it highlights for you to correct them.

The service is based on After the Deadline and is available for free. If you upgrade to 10.1 there is nothing to do to set it up. We’ve pre-package the connection to that service for you.

Tags:

Copyright © 2011 MindTouch, Inc. Powered by