Wrapping an APM implementation with Future<T>

In a previous post, I talked about implementing the Asynchronous Programming Model pattern using Future<T> from Parallel Extensions to the .NET Framework.  It’s also possible to go in the opposite direction, to create a Future<T> from an existing APM implementation.

As has been shown in previous examples, in this example we’ll take advantage of the “promise” capabilities provided by Future<T>.  This allows a Future<T> to be created without a Func<T> delegate, such that the future’s Value and Exception properties can be set explicitly, but only once; any thread waiting on the Future<T> will block until one of these properties is set.

static Future<T> Create<T>(
    Action<AsyncCallback> beginFunc,
    Func<IAsyncResult, T> endFunc)
{
    var f = Future<T>.Create();
    beginFunc(iar => {
        try { f.Value = endFunc(iar); }
        catch (Exception e) { f.Exception = e; }
    });
    return f;
}

This Create<T> method first creates a Future<T> as a promise.  It then calls a delegate provided by the caller to start the asynchronous action using the relevant BeginXx method; as an argument to that action, this code passes a delegate that will be called back by the asynchronous operation when it completes; the APM method will pass an IAsyncResult, which will then be passed in this callback to the endFunc delegate provided by the caller.  That endFunc will call the EndXx method, and will use the resulting value or exception to set the Future<T>.

As an example of using this, consider a FileStream.  FileStream exposes a BeginRead method:

IAsyncResult BeginRead(
    byte[] array, int offset, int numBytes,
    AsyncCallback
userCallback, object stateObject)

as well as a corresponding EndRead method:

int EndRead(IAsyncResult asyncResult);

If we want to create a Future<T> that represents an asynchronous read operation on a FileStream, we could do so with our new Create method as follows:

var readFuture = Create<int>(
   
ac => fs.BeginRead(buffer, 0, buffer.Length, ac, null),
   
fs.EndRead);

I can now go off and do something else, and when I later get readFuture’s Value, I’ll block until the asynchronous operation completes (or return immediately with the result if it’s already completed).  If the FileStream was created to support asynchronous I/O operations and if the underlying version of Windows supports asynchronous I/O, while the read request is happening, no threads will be used, and yet I’ll still have a valid Future<T> to represent the operation.

There are of course other possible designs for and implementations of such a Create method.  An alternate implementation might look like this:

static Future<T> Create<T>(
    IAsyncResult iar, Func<IAsyncResult, T> endFunc)
{
    var f = Future<T>.Create();
    ThreadPool.RegisterWaitForSingleObject(
            iar.AsyncWaitHandle, delegate
{
        try { f.Value = endFunc(iar); }
        catch (Exception e) { f.Exception = e; }
    }, null, -1, true);
    return f;
}

Here’s I’m using the same promise functionality of Future<T>, but instead of using the callback functionality of the APM, I’m relying on ThreadPool’s RegisterWaitForSingleObject functionality.  When the provided IAsyncResult’s AsyncWaitHandle is signaled (which will happen when the asynchronous operation completes), my delegate will be called to call endFunc and set the Future<T>’s Value or Exception property as shown previously.  This version can be used as follows:

var readFuture = Create<int>(
    fs.BeginRead(buffer, 0, buffer.Length, null, null),
    fs.EndRead);

Now comes the interesting question…

We don’t currently have a Create method like this on Future<T>.  You can write your own as has been shown here, but is this important enough for us to include one on Future<T> itself?  Do you have such a need in your applications?  Should we provide deeper support to further enable this kind of integration? Let us know 🙂