Call operations asynchronously with expiration

Question

I have a classic problem of sending messages asynchronously. Essentially, I need to send messages asynchronously, and then capture the response of the message when sending is complete. The problem is that I cannot figure out how to make one request cycle with expiration and short circuit.

Here is a sample template that I am currently using:

A specific delegate to activate

private delegate IResponse MessageDispatchDelegate(IRequest request); 

Sending Callback Messages

 var dispatcher = new MessageDispatchDelegate(DispatchMessage); dispatcher.BeginInvoke(requestMessage, DispatchMessageCallback, null); 

Message sending

 private IResponse DispatchMessage(IRequest request) { //Dispatch the message and throw exception if it times out } 

Receive send results as a response or exception

 private void DispatchMessageCallback(IAsyncResult ar) { //Get result from EndInvoke(r) which could be IResponse or a Timeout Exception } 

What I cannot understand is how to correctly implement the timeout / short circuit process in the DispatchMessage method. Any ideas would be appreciated.

Answer 1

After many scratches on my head, I was finally able to find a solution to my initial question. First of all, let me say that I have many excellent answers and I checked them all (commenting on each of the results). The main problems were that all the proposed solutions led to locks (leading to a 100% timeout scenario) or otherwise made the synchronous asynchronous process different. I don’t like answering my own question (the first time ever), but in this case I took advice on the StackOverflow FAQ, since I really learned my own lesson and wanted to share it with the community.

In the end, I combined the proposed solutions with calling delagates into alternative AppDomains. This is a bit more code, and it's a bit expensive, but it avoids the locks and allows for the completely asynchronous calls that I need. Here is a bit ...

First I needed something to call a delegate in another AppDomain

 /// <summary> /// Invokes actions in alternate AppDomains /// </summary> public static class DomainInvoker { /// <summary> /// Invokes the supplied delegate in a new AppDomain and then unloads when it is complete /// </summary> public static T ExecuteInNewDomain<T>(Delegate delegateToInvoke, params object[] args) { AppDomain invocationDomain = AppDomain.CreateDomain("DomainInvoker_" + delegateToInvoke.GetHashCode()); T returnValue = default(T); try { var context = new InvocationContext(delegateToInvoke, args); invocationDomain.DoCallBack(new CrossAppDomainDelegate(context.Invoke)); returnValue = (T)invocationDomain.GetData("InvocationResult_" + invocationDomain.FriendlyName); } finally { AppDomain.Unload(invocationDomain); } return returnValue; } [Serializable] internal sealed class InvocationContext { private Delegate _delegateToInvoke; private object[] _arguments; public InvocationContext(Delegate delegateToInvoke, object[] args) { _delegateToInvoke = delegateToInvoke; _arguments = args; } public void Invoke() { if (_delegateToInvoke != null) AppDomain.CurrentDomain.SetData("InvocationResult_" + AppDomain.CurrentDomain.FriendlyName, _delegateToInvoke.DynamicInvoke(_arguments)); } } } 

Second I needed something to organize the collection of the necessary parameters and collect / resolve the results. It will also determine the timeout and workflows that will be invoked asynchronously in the alternate AppDomain.

Note. In my tests, I extended the work dispatcher method to accept random time intervals to notice that everything works as expected in both timeouts and in the absence of a timeout.

 public delegate IResponse DispatchMessageWithTimeoutDelegate(IRequest request, int timeout = MessageDispatcher.DefaultTimeoutMs); [Serializable] public sealed class MessageDispatcher { public const int DefaultTimeoutMs = 500; /// <summary> /// Public method called on one more many threads to send a request with a timeout /// </summary> public IResponse SendRequest(IRequest request, int timeout) { var dispatcher = new DispatchMessageWithTimeoutDelegate(SendRequestWithTimeout); return DomainInvoker.ExecuteInNewDomain<Response>(dispatcher, request, timeout); } /// <summary> /// Worker method invoked by the <see cref="DomainInvoker.ExecuteInNewDomain<>"/> process /// </summary> private IResponse SendRequestWithTimeout(IRequest request, int timeout) { IResponse response = null; var dispatcher = new DispatchMessageDelegate(DispatchMessage); //Request Dispatch var asyncResult = dispatcher.BeginInvoke(request, null, null); //Wait for dispatch to complete or short-circuit if it takes too long if (!asyncResult.AsyncWaitHandle.WaitOne(timeout, false)) { /* Timeout action */ response = null; } else { /* Invoked call ended within the timeout period */ response = dispatcher.EndInvoke(asyncResult); } return response; } /// <summary> /// Worker method to do the actual dispatch work while being monitored for timeout /// </summary> private IResponse DispatchMessage(IRequest request) { /* Do real dispatch work here */ return new Response(); } } 

Third I need something worthwhile for the real thing, which asynchronously launches mailing lists

Note. This is just to demonstrate the asynchronous behavior I need. In fact, those First and Second elements demonstrate isolation of timeout behavior on alternative threads. This simply demonstrates how these resources are used.

 public delegate IResponse DispatchMessageDelegate(IRequest request); class Program { static int _responsesReceived; static void Main() { const int max = 500; for (int i = 0; i < max; i++) { SendRequest(new Request()); } while (_responsesReceived < max) { Thread.Sleep(5); } } static void SendRequest(IRequest request, int timeout = MessageDispatcher.DefaultTimeoutMs) { var dispatcher = new DispatchMessageWithTimeoutDelegate(SendRequestWithTimeout); dispatcher.BeginInvoke(request, timeout, SendMessageCallback, request); } static IResponse SendRequestWithTimeout(IRequest request, int timeout = MessageDispatcher.DefaultTimeoutMs) { var dispatcher = new MessageDispatcher(); return dispatcher.SendRequest(request, timeout); } static void SendMessageCallback(IAsyncResult ar) { var result = (AsyncResult)ar; var caller = (DispatchMessageWithTimeoutDelegate)result.AsyncDelegate; Response response; try { response = (Response)caller.EndInvoke(ar); } catch (Exception) { response = null; } Interlocked.Increment(ref _responsesReceived); } } 

In retrospect, this approach has some unintended consequences. Since the working method is found in an alternative AppDomain, this adds additional protection for exceptions (although this can also hide them), allows you to load and unload other managed assemblies (if required), and allows you to define highly restricted or specialized security contexts. This requires a bit more performance but provided a framework for answering my original question. Hope this helps someone.

Answer 2

  var dispatcher = new MessageDispatchDelegate(DispatchMessage); var asyncResult = dispatcher.BeginInvoke(requestMessage, DispatchMessageCallback, null); if (!asyncResult.AsyncWaitHandle.WaitOne(1000, false)) { /*Timeout action*/ } else { response = dispatcher.EndInvoke(asyncResult); }