1

I have a number of commands that make calls to a soap web service (Betfair API). All are of the classic asynchronous programming model type ...

public void DoXXX( <input parameters ...> )
{
    XXXRequest Request = new XXXRequest();
    // populate Request from input parameters ...
    BetfairService.BeginXXX( Request, XXXCallback, State );
}

private void XXXCallback(IAsyncResult Result)
{
    XXXResponse Response = BetfairService.EndXXX(Result);
    if (Response.ErrorCode == XXXErrorCode.OK)
        // store data from Response
    else
        // deal with error
}

I want to execute a specified set of commands, and then do some calculations using the combined returned data values, once all of commands are completed.

I'm able to do this as a sequence, by making a queue of commands and having each callback method trigger the next command in the queue once it's complete, with the calculation as the last item in the queue. This is relatively slow however.

My ideal solution would be to have all of these commands running in parallel and then to have the calculation triggered once all of the commands are completed. I've tried looking at Task.Factory.FromAsync(), but all of the examples I can find only include direct calls to BeginXXX / EndXXX, not doing anything with the response.

Does anyone have any pointers for a suitable solution to this problem?

Improve this question

3 Answers 3

Reset to default
2

To use FromAsync, you need to specify the return type:

var task = Task<XXXResponse>.Factory.FromAsync( ...

You then have a task with a Result property of type XXXResponse.

You could then use Parallel.Invoke to run the initial commands in parallel. This will block until all those tasks complete. Then you can do your "additional processing".

Or you could store the initial tasks in an array and use Task.Factory.ContinueWhenAll to create a continuation.

Nick

Improve this answer
Sign up to request clarification or add additional context in comments.

3 Comments

If I'm using Task.Factory.ContinueWhenAll, is there any way to force it to wait until all of the callbacks have actually finished processing rather than just having been called?
I'm not entirely sure what you're asking. The task returned from Task.Factory.FromAsync should not complete until the EndXXX method has completed. ContinueWhenAll also returns a Task object, which you can do all the normal task operations on, like Wait().
Actually that was my mistake. I'm developing for WPF, so everything that might have updated a bound property was wrapped in CurrentDispatcher.BeginInvoke() - this is what was causing my sequencing issues. This isn't required for tasks and all works fine. Thanks for the tip.
1

I would suggest that you look at Microsoft's Reactive Extensions (Rx) to do what you want. It lets you turn async operations (among other things) into observable LINQ queries.

Say I have these three functions that each take significant time to compute:

Func<int> fa = () =>
{
    Thread.Sleep(2000);
    return 42;
};

Func<int, string, string> fb = (n, t) =>
{
    Thread.Sleep(n * 1000);
    return t + n.ToString();
};

Func<DateTimeOffset> fc = () =>
{
    Thread.Sleep(1000);
    return DateTimeOffset.UtcNow;
};

I can then use the FromAsyncPattern method to turn these lambda functions into observable functions:

Func<IObservable<int>> ofa =
    Observable
        .FromAsyncPattern<int>(
            fa.BeginInvoke,
            fa.EndInvoke);

Func<int, string, IObservable<string>> ofb =
    Observable
        .FromAsyncPattern<int, string, string>(
            fb.BeginInvoke,
            fb.EndInvoke);

Func<IObservable<DateTimeOffset>> ofc =
    Observable
        .FromAsyncPattern<DateTimeOffset>(
            fc.BeginInvoke,
            fc.EndInvoke);

Now I can start all of the calls by simply doing this:

IObservable<int> oa = ofa();
IObservable<string> ob = ofb(1, "foo");
IObservable<DateTimeOffset> oc = ofc();

That effectively kicks off the three computations in parallel. Now we just have to bring the results together.

This is where LINQ comes in:

var query =
    from a in oa
    from b in ob
    from c in oc
    select new { a, b, c };

And I then subscribe to this query to get the results:

query.Subscribe(p =>
{
    Console.WriteLine(p.a);
    Console.WriteLine(p.b);
    Console.WriteLine(p.c);
});

In my testing I put timers around this code to compute actual execution time. Even though the total time should be 4 seconds if run in series this code finishes in 2 - the maximum time of any of the three.

Now this example is only a little facet of what Rx can do, but it is a good starting point.

Yell out if I can explain anything further.

Here are the links for Rx:

Improve this answer

Comments

0

You should have a counter of the executed service calls. In each callback method you should check this counter - if it equals the max number of service calls, you should do your additional processing, otherwise - you just increment the counter.

Improve this answer

Comments

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Start asking to get answers

Find the answer to your question by asking.

Ask question

Explore related questions

See similar questions with these tags.