A New Way to Approach APM in .NET

• Download ApmHelperExample.zip - 19.77 KB
• Download just the CS file of the ApmHelperBase that supports timeout - 12.75

Introduction

Threading is a major part of application development.  .NET offers some great ways to handle threading, one being the Asynchronous Programming Model (APM). When using APM though, you tend to write a lot of boilerplate and repetitive code - particularly in a WinForms environment.  The code I am introducing wraps that boilerplate code and abstracts some of the responsibilities a developer might have when marshalling callbacks to the appropriate thread.

Background

APM exposes a number of different models to use when threading (callbacks, polling, blocking).  My APM helper is principally designed to handle the callback model. Typically you cache an IAsycnResult from a BeginInvoke() operation and register a callback. If you offer the ability to invoke again while the previous BeginInvoke() has yet to complete, often times you want to discard the results from the first BeginInvoke(). This is simple to do by checking the IAsyncResult you cached and doing a reference check with it in the callback.  Should they refer to the same memory location, you have the correct callback and, therefore, the correct results. If you are in a WinForms environment, you usually will do something with this data in the UI. Of course, your callback is executing on the background worker so if you don't marshall the data to the UI thread, then you get a lovely InvalidOperationException "Cross-thread operation not valid: Control 'blahblahblah' accessed from a thread other than the thread it was created on."  The same goes for if you catch an exception in the callback and are not careful to marshall it properly to the UI thread. 

Recommended APM Usage 

using System;
using System.Runtime.Remoting.Messaging;
using System.Windows.Forms;

namespace ApmHelperWinformsExample
{
    /// <summary>
    /// The ordained way of programming apm
    /// </summary>
    public partial class OrdainedApm : Form
    {
        private delegate double Add3(int a, int b, int c);
        private readonly object _lock = new object();
        private IAsyncResult _current;

        public OrdainedApm()
        {
            InitializeComponent();
        }

        private IAsyncResult Current
        {
            get { lock (this._lock) return this._current; }
            set { lock (this._lock) this._current = value; }
        }
        
        private void button1_Click(object sender, EventArgs e)
        {
            int a = int.Parse(this.textBox1.Text);
            int b = int.Parse(this.textBox2.Text);
            int c = int.Parse(this.textBox3.Text);
            Add3 add3 = SomeWebServiceMaybe.SomeLongRunningAddFunction;
            this.Current = add3.BeginInvoke(a, b, c, this.Callback, null);
        }

        private void Callback(IAsyncResult result)
        {
            try
            {
                var asyncResult = (AsyncResult) result;
                //ReturnMessage msg = (ReturnMessage)asyncResult.GetReplyMessage();
                var add3Del = (Add3) asyncResult.AsyncDelegate;
                double d = add3Del.EndInvoke(result);
                if(result == this.Current)
                {
                    this.Display(d);
                }
            }
            catch(Exception ex)
            {
                if(result == this.Current)
                {
                    this.Display(ex);
                }
            }
        }

        private void Display(double d)
        {
            if (this.InvokeRequired)
            {
                this.Invoke((MethodInvoker) (() => this.Display(d)));
                return;
            }
            this.label1.Text = d.ToString();
        }

        private void Display(Exception ex)
        {
            if (this.InvokeRequired)
            {
                this.Invoke((MethodInvoker)(() => this.Display(ex)));
                return;
            }
            this.label1.Text = ex.ToString();
            
        }
    }
}

As you can see, there is a lot of boilerplate code here. Declaring a delegate, caching and providing thread-safe access to the IAsyncResult, try/catching EndInvoke(), casting IAsyncResult to AsyncResult to grab the delegate to make the EndInvoke() call (sure this can be passed in as state, but it is already provided when using .NET's delegates), and Invoking on the UI Thread after we get our data. 

I Would Prefer to Write Something Like

using System;
using System.Windows.Forms;

namespace ApmHelperWinformsExample
{
    /// <summary>
    /// the way I want to program apm
    /// </summary>
    public partial class ApmHelperExample : Form
    {
        private readonly ApmHelper<int, int, int, double> _helper;

        public ApmHelperExample()
        {
            InitializeComponent();
            this._helper = new ApmHelper<int, int, int, double>
			(SomeWebServiceMaybe.SomeLongRunningAddFunction);
        }

        private void Callback(double result)
        {
            this.label1.Text = result.ToString();
        }

        private void ExceptionHandler(Exception ex)
        {
            this.label1.Text = ex.ToString();
        }

        private void button1_Click(object sender, EventArgs e)
        {
            int a = int.Parse(this.textBox1.Text);
            int b = int.Parse(this.textBox2.Text);
            int c = int.Parse(this.textBox3.Text);
            this._helper.InvokeAsync(a, b, c, this.Callback, this.ExceptionHandler);
        }
    }
}

Notice we don't need to cache any results from the InvokeAsync(), we don't need to worry about marshalling any data to UI thread, and we don't need to worry about catching any exceptions and marshalling those to the UI thread. All we need to provide are the arguments (note these are strongly-typed as well), the callback, and optionally, an exception handler. 

The Code for the APMHelper

public enum ThreadCallback
{
    /// <summary>
    /// This is the thread that InvokeAsync() is called on
    /// </summary>
    AsyncCallerThread,
    /// <summary>
    /// This is the thread the InvokeAsync() spawned
    /// </summary>
    WorkerThread
}

/// <summary>
/// This class is responsible for tracking the IAsyncResult from a BeginInvoke(),
/// Catching exceptions from EndInvoke(), and Marshalling the data or exception to the
/// correct thread
/// </summary>
/// <typeparam name="TResult">result of function call</typeparam>
public abstract class ApmHelperBase<TResult> : IDisposable
{
    /// <summary>
    /// lock to preserve thread safety on getting/setting IAsyncResult
    /// </summary>
    private readonly object _lock = new object();
    /// <summary>
    /// Pointer to method to execute when EndInvoke() throws exception
    /// </summary>
    private readonly Action<Exception> _defaultExceptionHandler;
    /// <summary>
    /// Method to get begin, end invoke functions from
    /// </summary>
    private readonly Delegate _function;
    /// <summary>
    /// cache of BeginInvoke method
    /// </summary>
    private readonly MethodInfo _beginInvokeMethod;
    /// <summary>
    /// cache of EndInvoke method
    /// </summary>
    private readonly MethodInfo _endInvokeMethod;
    /// <summary>
    /// cache of callback that all BeginInvoke()s wire to
    /// </summary>
    private readonly AsyncCallback _asyncCallback;
    /// <summary>
    /// Cache of the current call from BeginInvoke()
    /// </summary>
    private IAsyncResult _current;

    /// <summary>
    /// ctor
    /// </summary>
    protected ApmHelperBase(Delegate function, Action<Exception> exceptionHandler)
        : this(function, exceptionHandler, ThreadCallback.AsyncCallerThread) { }

    /// <summary>
    /// ctor
    /// </summary>
    /// <param name="function">function user is going to bind to.
    /// This will actually be some func but we don't care here</param>
    /// <param name="exceptionHandler">optional method user wants 
    /// to be notified if async call throws exception</param>
    /// <param name="threadCallback">Which thread should callbacks occur on</param>
    protected ApmHelperBase(Delegate function, 
		Action<Exception> exceptionHandler, ThreadCallback threadCallback)
    {
        if (null == function) throw new ArgumentNullException("function");
        this._function = function;
        // cache the methods
        Type type = function.GetType();
        this._beginInvokeMethod = type.GetMethod("BeginInvoke");
        this._endInvokeMethod = type.GetMethod("EndInvoke");
        // if no ex handler, we use our own
        this._defaultExceptionHandler = exceptionHandler ?? DefaultExceptionHandler;
        // all async calls will get pointed to this callback
        this._asyncCallback = this.Callback;
        // cache which thread user wants callbacks on..he can change later if he wants
        this.TheThreadCallback = threadCallback;
    }

    /// <summary>
    /// User can set if they want callbacks to occur on the 
    /// thread this object is called InvokeAsycn(), or on the thread
    /// spawned from the BeingInvoke() method.
    /// </summary>
    /// <remarks>This applies to the exception handler as well</remarks>
    public ThreadCallback TheThreadCallback { get; set; }

    /// <summary>
    /// Provides the asyncresult in a threadsafe manner
    /// </summary>
    public IAsyncResult CurrentIAsyncResult
    {
        get { lock (this._lock) return this._current; }
        private set { lock (this._lock) this._current = value; }
    }

    /// <summary>
    /// Sets IssueCallbacksOnInvokesAsync to false and
    /// wipes out the CurrentIAsyncResult so no callback fires
    /// </summary>
    /// <remarks>Subsequent calls to InvokeAsync() will
    /// set the CurrentIAsyncResult so the last InvokeAsync() 
    /// will get the callback</remarks>
    public void Ignore()
    {
        this.IssueCallbacksOnInvokesAsync = false;
        this.CurrentIAsyncResult = null;
    }

    /// <summary>
    /// If true, APMHelper will issue callbacks on ALL BeginInvokes(), otherwise
    /// only the last InvokeAsync() gets the callback
    /// </summary>
    public bool IssueCallbacksOnInvokesAsync { get; set; }

    /// <summary>
    /// Ignores all outstanding calls.
    /// </summary>
    /// <remarks>You could actually start using it again</remarks>
    public void Dispose()
    {
        this.Ignore();
    }

    /// <summary>
    /// User should convert his arguments in order into args parm
    /// </summary>
    /// <param name="args">T1, T2..etc</param>
    /// <param name="userCallback">method user wants results pumped to</param>
    /// <param name="exceptionHandler">method user wants exceptions pumped into</param>
    protected void InvokeAsync(List<object> args, 
	Action<TResult> userCallback, Action<Exception> exceptionHandler)
    {
        // if a sync context is available and user wants callback on AsyncCallerThread,
        // then callback will happen on the thread calling this method now.
        // Otherwise, the normal bg thread will call the callback
        args.Add(
            this.TheThreadCallback == ThreadCallback.AsyncCallerThread
                ? SyncContextAsyncCallback.Wrap(this._asyncCallback)
                : this._asyncCallback);
        // we need to pass in the pointer to the method the user wants his notification
        args.Add(new CallbackState(userCallback, exceptionHandler));
        // even though we call Invoke, 
        // we are actually calling the BeginInvoke() so this won't block
        this.CurrentIAsyncResult = 
	(IAsyncResult) this._beginInvokeMethod.Invoke(this._function, args.ToArray());
    }

    /// <summary>
    /// User should convert his arguments in order into args parm
    /// </summary>
    /// <param name="args">T1, T2..etc</param>
    /// <param name="userCallback">method user wants results pumped to</param>
    protected void InvokeAsync(List<object> args, Action<TResult> userCallback)
    {
        this.InvokeAsync(args, userCallback, this._defaultExceptionHandler);
    }

    /// <summary>
    /// all async calls come through here to make sure they are valid
    /// </summary>
    private void Callback(IAsyncResult result)
    {
        bool correctCall = result == this.CurrentIAsyncResult || 
					this.IssueCallbacksOnInvokesAsync;
        var tmp = (AsyncResult) result;
        var callbackState = (CallbackState)((AsyncResult)result).AsyncState;

        TResult output;
        try
        {
            // get our results
            output = (TResult)this._endInvokeMethod.Invoke
				(this._function, new[] { result });
        }
        catch (Exception ex)
        {
            if (correctCall)
            {
                // get our callback
                ExecuteExceptionHandler(ex, callbackState.ExceptionHandler);
            }
            return;
        }
        if (!correctCall) return;
        if (null == callbackState.UserCallback) return; // user might have just 
						// issued fire and forget
        // notify the user
        callbackState.UserCallback(output);
    }

    private static void ExecuteExceptionHandler
	(Exception exception, Action<Exception> exHandler)
    {
        if (null != exHandler) exHandler(exception);
    }

    private static void DefaultExceptionHandler(Exception ex)
    {
        // log if you want
        throw ex;
    }

    private sealed class CallbackState
    {
        public readonly Action<TResult> UserCallback;
        public readonly Action<Exception> ExceptionHandler;
        public CallbackState(Action<TResult> userCallback, 
				Action<Exception> exceptionHandler)
        {
            this.UserCallback = userCallback;
            this.ExceptionHandler = exceptionHandler;
        }
    }
}

Points of Interest

Essentially, you can see that subclasses (see below for an example of one) provide the delegate which is passed into the constructor. We then use reflection to grab the begin and endinvoke methods. The action happens in InvokeAsync(). This method will invoke the BeginInvoke() with the arguments and add two more arguments: the callback and the state. The state will usually be the Action<TResult> method that should get called when the EndIvoke() completes. Notice the call to SyncContextAsyncCallback.Wrap(this._asyncCallback). This code I actually got from Jeffrey Richter and he provides it in his Power Threading Library. What this code is doing is basically caching the SynchronizationContext used from the InvokeAsync() call. In the example above, this would be the UI thread. The great thing about this, is that it will marshall the results or exception to the thread that invoked InvokeAsync(). What happens is my callback is wrapped in the instance of SyncContextAsyncCallback's callback. And that callback is responsible for posting to the SynchronizationContext of the InvokeAysnc() call, the callback registered in APMHelperBase. Of course, if there is no SynchronizationContext or the user has specified the callback to be issued on the background worker via the ThreadCallback property, then we ignore this process. 

APMHelperBase is abstract so here is an example of a subclass. The purpose of this subclass, and one for all the Func<> signatures, is to provide the strong-typing.

/// <summary>
/// Provides async service for Func T1, T2, T3, TResult
/// </summary>
public class ApmHelper<T1, T2, T3, TResult> : ApmHelperBase<TResult>
{
    /// <summary>
    /// ctor
    /// </summary>
    public ApmHelper(Func<T1, T2, T3, TResult> func) : this(func, null) { }

    /// <summary>
    /// ctor
    /// </summary>
    public ApmHelper(Func<T1, T2, T3, TResult> func, Action<Exception> exceptionHandler)
        : this(func, exceptionHandler, ThreadCallback.AsyncCallerThread)
    {
    }

    /// <summary>
    /// ctor
    /// </summary>
    public ApmHelper(Func<T1, T2, T3, TResult> func, ThreadCallback threadCallback)
        : this(func, null, threadCallback)
    {
    }

    /// <summary>
    /// ctor
    /// </summary>
    public ApmHelper(Func<T1, T2, T3, TResult> func, 
		Action<Exception> exceptionHandler, ThreadCallback threadCallback)
        : base(func, exceptionHandler, threadCallback)
    {
    }

    /// <summary>
    /// starts the async call
    /// </summary>
    public void InvokeAsync(T1 t1, T2 t2, T3 t3, Action<TResult> callback)
    {
        this.InvokeAsync(new List<object> { t1, t2, t3 }, callback);
    }

    /// <summary>
    /// starts the async call
    /// </summary>
    public void InvokeAsync(T1 t1, T2 t2, T3 t3, 
	Action<TResult> callback, Action<Exception> exceptionHandler)
    {
        this.InvokeAsync(new List<object> { t1, t2, t3 }, callback, exceptionHandler);
    }
}

A Couple of Things

There are a few options you can set.  For instance, if you want to get every callback from InvokeAsync(), you can set IssueCallbacksOnInvokesAsync = true. However, you will get them in order of the completion of the calls, not necessarily the order you executed.  Ignore() will basically set the current IAsycnResult to null so we don't issue our callback. This is useful if you are closing a form and still have an outstanding call executing.  And should you want to get your hands on the current IAsyncResult, you can and then of course you can do your polling, blocking, etc. if you wanted to.