Implementing the Asynchronous Programming Model with Future
One of our design goals for the Task Parallel Library is to integrate well into existing asynchronous mechanisms in the .NET Framework. And one of the most common concurrency-related patterns in the .NET Framework is the Asynchronous Programming Model (APM), which typically manifests as a BeginXx method that kicks off an asynchronous operation and returns an IAsyncResult, along with an EndXx method that accepts an IAsyncResult and returns the computed value.
Implementing this for computationally intensive asynchronous operations can be done with System.Threading.Tasks.Future<T>, as Future<T> derives from System.Threading.Tasks.Task, and Task implements IAsyncResult. Imagine you had a class used for computing the value of pi to an arbitrary number of decimal places:
public class PI
{
public string Calculate(int decimalPlaces) { … }
}
We can add BeginCalculate and EndCalculate methods that provide a trimmed-down implementation of the APM for Calculate with only a few lines of code:
public class PI
{
…
public IAsyncResult BeginCalculate(int decimalPlaces)
{
return Future.Create(() => Calculate(decimalPlaces));
}
public string EndCalculate(IAsyncResult ar)
{
var f = ar as Future<string>;
if (f == null) throw new ArgumentException(“ar”);
return f.Value;
}
}
However, the full APM pattern dictates that BeginXx should also accept an AsyncCallback that is invoked when the operation completes, along with some user-defined state. The AsyncCallback addition is easy to implement with the current design of Future<T>:
public IAsyncResult BeginCalculate(
int decimalPlaces, AsyncCallback ac)
{
var f = Future.Create(() => Calculate(decimalPlaces));
if (ac != null) f.Completed += delegate { ac(f); };
return f;
}
I simply register with the Future<T>’s Completed event such that when the Future<T> completes, it invokes the AsyncCallback. Easy enough. Unfortunately, with the current design of Future<T>, supporting the state argument is more difficult, since unlike Task, none of the factories for Future<T> accept an arbitrary state argument (this is something we’re planning to rectify). For now with the current CTP bits, you can work around this by creating your own small wrapper for the Future<T> and the state object:
private class FutureWithState<T> : IAsyncResult
{
public IAsyncResult Future;
public object State;
public object AsyncState { get { return State; } }
public WaitHandle AsyncWaitHandle {
get { return Future.AsyncWaitHandle; } }
public bool CompletedSynchronously {
get { return Future.CompletedSynchronously; } }
public bool IsCompleted {
get { return Future.IsCompleted; } }
}
The modifications to our implementations of BeginCalculate and EndCalculate to support state are then very straightforward:
public IAsyncResult BeginCalculate(
int decimalPlaces, AsyncCallback ac, object state)
{
var f = Future.Create(() => Calculate(decimalPlaces));
var fws = new FutureWithState<string> {
Future = f, State = state };
if (ac != null) f.Completed += delegate { ac(fws); };
return fws;
}public string EndCalculate(IAsyncResult ar)
{
var f = ar as FutureWithState<string>;
if (f == null) throw new ArgumentException(“ar”);
return f.Future.Value;
}
This same container approach can work for scenarios where you want to store more arbitrary state with your IAsyncResult.
Note, too, that Future<T> can also be used for situations where the asynchronous work isn’t computationally intensive. Consider the System.Net.NetworkInformation.Ping class that was introduced in the .NET Framework 2.0. Rather than following the APM pattern, Ping derives from Component and follows the Event-based Asynchronous Pattern (EAP). So rather than exposing BeginSend and EndSend methods that deal with IAsyncResults, it exposes a SendAsync method that returns void, along with a PingCompleted event that is raised when the asynchronous ping send operation completes. What if we wanted to use asynchronous pings, but we wanted to expose them through the APM pattern. We could easily implement this with Future<T> as follows:
public class MyPing
{
public IAsyncResult BeginPing(string host)
{
return Future.Create(() => new Ping().Send(host));
}
public PingReply EndPing(IAsyncResult ar)
{
var f = ar as Future<PingReply>;
if (f == null) throw new ArgumentException(“ar”);
return f.Value;
}
}
but this has some serious downsides: foremost, the implementation of Send is not computationally intensive, and yet the Future<T> is going to be using and blocking a thread to do a synchronous send. What we really want to do is take full advantage of Ping’s asynchronous send implementation. And we can do that by using Future<T>’s “promise” capabilities:
public class MyPing
{
public IAsyncResult BeginPing(string host, AsyncCallback ac)
{
var f = Future<PingReply>.Create();
if (ac != null) f.Completed += delegate { ac(f); };
Ping p = new Ping();
p.PingCompleted += (sender, state) =>
{
if (state.Error != null) f.Exception = state.Error;
else f.Value = state.Reply;
};
p.SendAsync(host, null);
return f;
}
public PingReply EndPing(IAsyncResult ar)
{
var f = ar as Future<PingReply>;
if (f == null) throw new ArgumentException(“ar”);
return f.Value;
}
}
The Future<T> I create here has no delegate associated with it. Instead, it allows its Value and Exception properties to be set (only once), and someone waiting on Future<T>’s Value will block until either Exception or Value is set. I take advantage of that by setting these properties in the PingCompleted event handler. With very little code, I now can get a fully-working IAsyncResult that represents an asynchronous ping, where the original asynchronous behavior was exposed through a different mechanism.
Light
Dark
0 comments