Justin Rogers

Generic, Cancellable, Asynchronous operations? Yeah, I'll blog about that.

Just code today my friends.  The design is simple, create a framework that allows running an asynchronous delegate on the thread pool in a manner where it can be cancelled and any exceptions can be handled.  We use the same BeginInvoke/EndInvoke model as the rest of the asynchronous programming models under the framework.  Our generic classes allow for the creation of a new generic asynchronous result object for each piece of work we want to do.  To fully define a generic async operation we need to specify some from of input, the type of the output, and we can even strongly type the async state object.

ar = GenericCancellableOperation<RegexInput, int, Match>.BeginInvoke(
    new RegexInput(regex, "555"),
    new GenericWorkDelegate<Match, RegexInput>(RegexWorkDelegate),
    new AsyncCallback(MatchComplete),
    counter++
);

The above is a sample usage that does the same thing as my previously defined asynchronous regular expression runner.  We can no longer specify entire strings of input parameters, so we have to wrap them all in a single object.  Here I've used RegexInput to define the Regex and input to match against.  We still use the old AsyncCallback model, even though we could strongly type that as well.  In fact, BeginInvoke and EndInvoke still use IAsyncResult instead of more strongly typed equivalents.  The GenericWorkDelegate will always match the input and return types for the operation generic, basically allowing us to delegate whatever work needs to occur back to your code.  It gets wrapped with all of the protective stuff automatically.  You still can't run arbitrary code, because any old fool could wrap a try...catch in their delegate and handle the ThreadAbortException that would normally terminate the operation.

using System;
using System.Threading;

public delegate ReturnType GenericWorkDelegate<ReturnType, InputType>(InputType input);

public class GenericCancellableOperation<InputType, AsyncStateType, ReturnType> {
    public class GenericAsynchronousResult : IAsyncResult {
        // Generic State Parameters
        internal AsyncStateType asyncState = AsyncStateType.default;
        internal InputType input = InputType.default;
        internal ReturnType output = ReturnType.default;
        internal GenericWorkDelegate<ReturnType, InputType> workCallback = GenericWorkDelegate<ReturnType, InputType>.default;
       
        // Non generic state
        internal bool complete = false;
        internal bool completedSynchronously = false;
        internal bool cancelled = false;
        internal ManualResetEvent waitHandle = null;
        internal AsyncCallback callback = null;
        internal Thread currentThread = null;
        internal Exception innerException = null;
        internal Object lockObject = new Object();
       
        internal GenericAsynchronousResult(InputType input, GenericWorkDelegate<ReturnType, InputType> workCallback, AsyncCallback callback, AsyncStateType state) {
            this.asyncState = state;
            this.input = input;
            this.callback = callback;
            this.workCallback = workCallback;
            this.waitHandle = new ManualResetEvent(false);
        }

        public object AsyncState { get { return this.asyncState; } }
        public bool CompletedSynchronously { get { return this.completedSynchronously; } }
        public bool IsCompleted { get { return this.complete; } }
        public WaitHandle AsyncWaitHandle { get { return this.waitHandle; } }
       
        public AsyncStateType TrueAsyncState { get { return this.asyncState; } }
        public InputType Input { get { return this.input; } }
        public ReturnType Output { get { return this.output; } }
       
        public bool Cancel() {
            lock(lockObject) {
                if ( this.cancelled || this.complete ) {
                    return false;
                }
               
                this.cancelled = true;
                if ( this.currentThread != null ) {
                    this.currentThread.Abort();
                }
            }

            this.waitHandle.Set();
            return true;
        }
       
        internal void Complete() {
            lock(lockObject) {
                if ( this.complete || this.cancelled ) {
                    return;
                }
               
                this.complete = true;
            }
           
           
            this.waitHandle.Set();
            if ( callback != null ) {
                callback(this);
            }
        }
    }

    public static GenericAsynchronousResult BeginInvoke(
        InputType input,
        GenericWorkDelegate<ReturnType, InputType> workCallback,
        AsyncCallback callback,
        AsyncStateType state)
    {
        GenericAsynchronousResult result =  new GenericAsynchronousResult(input, workCallback, callback, state);
        if ( !result.complete ) {
            ThreadPool.QueueUserWorkItem(new WaitCallback(ThreadPool_WaitCallback), result);
        }
       
        return result;
    }
   
    private static void ThreadPool_WaitCallback(object state) {
        GenericAsynchronousResult result = state as GenericAsynchronousResult;
       
        if ( result != null && !result.cancelled ) {
            result.currentThread = Thread.CurrentThread;

            try {
                result.output = result.workCallback(result.input);
            } catch (ThreadAbortException) {
                Thread.ResetAbort();
            } catch (Exception exc) {
                result.innerException = exc;
            } finally {
                result.currentThread = null;
                result.Complete();
            }
        }
    }
   
    public static ReturnType EndInvoke(IAsyncResult ar) {
        GenericAsynchronousResult result = ar as GenericAsynchronousResult;
       
        if ( result != null ) {
            result.AsyncWaitHandle.WaitOne();
            if ( result.innerException != null ) {
                throw result.innerException;
            }
            return result.output;
        } else {
            throw new Exception("EndInvoke was called with the wrong async result");
        }
    }
}

I still don't find this as useful as the regex only version.  I think adapting the code to the particular asynchronous operation you are working on is most sufficient.  In the case of regular expressions, you could even embed the ability to abort the problem within the lexical scanner, thus removing the necessity of thread aborts and exceptions.  In fact for something like an interpreter aborting as an embedded instruction is the best way to go about things.  For now, we don't have the ability to modify the construction of regular expressions under .NET and we can't embed such abort instructions, so we'll have to live with little hacks.