How to properly join results of multiple async methods?

Question

In Silverlight app I need to run multiple async calls in parallel - they will call server to check data, etc. My pseudo-code-setup looks like this

public interface IAsyncVerifier
{
    event EventHandler OnCompleted;

    void Verify();
}

// Subscribe to verifier completion events
foreach (var verifier in this.VerifierEngine.GetVerifiers())
{
    var asyncVerifier = verifier as IAsyncVerifier;
    if (asyncVerifier == null || !asyncVerifier.IsVerifierRunning()) continue;

    asyncVerifier.OnCompleted += (s, a) => DoWhat(); // ????????
    asyncVerifier.Verify();
}

So, in my mind I have interface common for all those tasks. Then before execution I subscribe to all of their events and ... ?

I need to somehow wait for them to complete and proceed with what I'm doing. How to write remaining code properly? I see some kind of counter that I increment every time I call Verify() and decrement every time I get OnCompleted. But I don't get whole picture on how to wrap it together so it doesn't smell. Also, Silverlight runs those on background threads if I understand correctly (calls to WCF services)

EDIT:

I'm working with 3rd party verification engine. this.Verifier.VerifierEngine is not something I wrote. I can only add verifiers to collection. Usually they execute on main thread to validate properties, etc. My need is to create verifier that makes call back to server, hence async in Silverlight. I'm not dealing with threads manually, when I make call to service it returns on main thread so it's fine, this part hidden from me. All I want is to be able to "wait" until all of the verifiers completed. The only way I can check right now is the way I'm doing.

Therefore my UI is quirky. User hits' save after done modifying field and my verification starts. I check if all verifiers IsVerifierRunning and give user message that he needs to retry. I really need to be able to wait up for all to complete and then proceed.

1. I don't need to worry about threads unless proper solution will involve working with threads
2. I have access to results already

Answer 1

I too suggest using the Reactive Framework (Rx) to do what you want.

First up you need a function that will turn IAsyncVerifier into an observable that will kick off the verification process and subscribe to the OnCompleted. Here it is:

Func<IAsyncVerifier, IObservable<Unit>> startVerifier = av =>
    Observable.Create<Unit>(o =>
    {
        var subject = new AsyncSubject<Unit>();
        var s1 =
            Observable
                .FromEventPattern(
                    h => av.OnCompleted += h,
                    h => av.OnCompleted -= h)
                .Select(_ => Unit.Default)
                .Subscribe(subject);
        var s2 = subject.Subscribe(o);
        av.Verify();
        return new CompositeDisposable(s1, s2);
    });

This function uses an AsyncSubject to make sure that it captures the event firing.

Now the query becomes quite straight forward.

var query = (
    from av in VerifierEngine
        .GetVerifiers()
        .OfType<IAsyncVerifier>()
        .ToObservable()
    where !av.IsVerifierRunning()
    from result in startVerifier(av)
    select av)
        .ToArray();

In the Rx world, since this is a SelectMany query the default is to run these on the thread pool - so it is run in parallel.

The final .ToArray() call is a little different to the enumerable version of the function. The observable version changes an IObservable<T> (an observable that returns zero or more values) into IObservable<T[]> (an observable that returns a single array of zero or more values). In other words, this will wait until all of the verifiers are finished before returning them all.

Now you just have to subscribe to the query and run whatever code you want to knowing that all the verifiers have completed.

query.Subscribe(avs => DoWhat());

If you want to run some code each time a verifier finishes and when all of them finish then you can take off the .ToArray() call and subscribe to the query like so:

query.Subscribe(av => { /* Each */ }, () => { /* All */ });

I hope this is all clear enough.

Answer 2

From what I've heard about your requirements, I think a simple counter is sufficient. For now, I'm going with the solution that is closest to what you posted and listing the assumptions that need to be valid in order for it to work. If any assumptions are invalid, let me know and I'll update the code appropriately.

If your loop is running on the UI thread it can be as follows. There are several assumptions in here. First is that we are on the UI thread and the calls to verify execute on the ui thread. This puts the DispatcherSynchronizationContext into effect. Completion events from wcf will execute in the same SynchronizationContext that they began in. This will effectively serialize their execution which will give your code a very single threaded behavior, so locking/etc isn't necessary on the counter. Some more on that here http://msdn.microsoft.com/en-us/library/z8chs7ft(v=vs.95).aspx . Second assumption is that the verify calls will always complete, even if they error.

Edited to match unsubscribe requirement. Bear with any typo's as I just used notepad for this. EventHandler may have to be a specific type of event handler depending on your event args.

int activeVerifiers = 0;
List<VerificationCleanup> cleanups = new List<VerificationCleanup>();

EventHandler completionHandler = (s, a) => 
{
     activeVerifiers--;
     if(activeVerifiers == 0)
     {
          //execute some sort of message/notification/all complete event

          foreach(VerificationCleanup cleanup in cleanups)
          {
            cleanup.Cleanup();
          }
     }
}; 

foreach (var verifier in this.VerifierEngine.GetVerifiers())
{
    var asyncVerifier = verifier as IAsyncVerifier;
    if (asyncVerifier == null || !asyncVerifier.IsVerifierRunning()) continue;

    cleanups.Add(new VerificationCleanup(completionHandler, asyncVerifier));

    activeVerifiers++;
    asyncVerifier.OnCompleted += completionHandler
    asyncVerifier.Verify();
}

private class VerificationCleanup
{
    private EventHandler eventHandler = null;
    private IAsyncVerifier verifier = null;

    public VerificationCleanup(EventHandler eventHandler, IAsyncVerifier verifier)
    {
        this.eventHandler = eventHandler;
        this.verifier = verifier;
    }

    public void Cleanup()
    {
        if(this.verifier != null)
        {
            this.verifier.OnCompleted -= this.eventHandler;
        }
    }
}

Answer 3

I use Linq to Rx using System.Reactive.Linq; Observable.FromEvent Zip the method

  var InitNetCellCache = InitNetCell();
            InitNetCellCache.Subscribe((s) =>
            {
                Debug.WriteLine("init NetCell Cache OK.");

                //ScheduleCrrentThread(innerAction);
            });
            var InitKPIMapCache = InitKPIMapCell();
            var zipped = InitKPIMapCache.Zip(InitNetCellCache, (left, right) => left + " - " + right);

            zipped.Subscribe((s) =>
                {
                    //When all async methods are completed
                    runtime.Show();
                }
            );

  IObservable<object> InitNetCell()
    {
        IUnityContainer unityContainer = ServiceLocator.Current.GetInstance<IUnityContainer>();
        INetCellManager inc = unityContainer.Resolve<INetCellManager>();
        var NetCell = Observable.FromEvent<InitCacheCompletedDelegate, object>(
         h => ((object sendera) => h.Invoke(sendera)),
         h => inc.InitCacheCompleted += h,
         h => inc.InitCacheCompleted -= h);
         //Run you async method
        inc.InitCache();
        return from e in NetCell select e;
    }

Answer 4

Silverlight does have the ManualResetEvent, which would give you what you need. However, the code would need to look different than what you have.

Is there a particular reason that you wouldn't want to use a counter and a spin loop? It isn't as pretty, but it also wouldn't require as many code changes as other approaches.

Erick