Parallel execution of async calls

Question

Hi,

lI have a List of IObservable<T> (each item is representing an async call) and I want to run them in parallel. I came up with this solutions (which seems to work) but I'm not sure if it really works in parallel:

public IObservable<T> Parallel<T>(IEnumerable<IObservable<T>> asyncCalls)
{
    return
        asyncCalls.ToObservable()
        .SelectMany(x => x);
}

Is this the right implementation?

Best regards and thanks,
  forki

Answer 1

Probably you want to use ForkJoin.

Answer 2

Hi,

the ForkJoin seems to give me a IObservable<TSource[]> - which means I'll be notified if all async calls are ready.
What I really want is to start all async calls and whenever one is ready I want to get the result.

Maybe I just have to subscribe to all observables in the list?!
  But if I do so, they are all processed on the same thread.

Regards,
  forki

Answer 3

A nice scenario for this could be a parallel picture downloader.

Whenever a picture is downloaded I want to show it in my UI.
Of course some pictures are smaller than others so normally they would appear earlier on my screen.

I saw this sceario in Luke Hobans PDC talk on F#, but unfortunately he used Async.StartAsTask which is not available in F# for .NET 3.5.

Answer 4

Sounds like you're looking for the Amb operator (see my blog post: http://blogs.msdn.com/jeffva/archive/2009/11/18/amb-materialize-and-dematerialize.aspx)

Regards,

Jeffrey

Answer 5

Hi Jeffrey,

I don't really get how this could work with Amb.

I have an array with URLs for pictures and I want to download them all (in parallel). Everytime an image is ready I want to show it on the screen.

Regards,
  Steffen

Answer 6

You can use the operator Merge

return asyncCalls.Merge();

Answer 7

Hi Steffen,

Merge seems to be the right one ;-)

I put it all together into a blog post .

Regards,
  forki

Answer 8

This was a very interesting question, but when I tried it I didn't get what I expected. In the code below I'm trying to simulate some long-running calculation where I launch several such calculations in parallel and then collect the results at the end.

void Run()
{
  var xs = new int[] { 1, 2, 3, 4, 5, 6, 7, 8, 9, 10 };

  xs
    .Select((a) => Observable.Return(Calculate(a)))
    .Merge()
    .Subscribe((a) => Console.WriteLine(a));

  Console.ReadKey();
}

int Calculate(int i)
{
  Console.WriteLine("Starting calculation " + i);
  Thread.Sleep(5000);
  Console.WriteLine("Done");
  return i;
}

However, when I ran this code each calculation happens in series. I tried using Observable.Return(Scheduler.Later, Calculate(a)) but that didn't help. I also tried using ForkJoin rather than Merge but that didn't help either. What am I doing wrong?

Answer 9

Hint: Replace Observable.Return with Observable.Start or create a Task.ToObservable() with the Pfx.

Answer 10

Observable.Start worked. Thanks!

Answer 11

Hi !

Could you please be so kind and show me the code please?
I am just really very new to Rx and may be, I am missing some other expertise [in me ;-) ].
If I use your original sample, it runs, but if I use ".Start(..)" instead, VS tells me, that I have to pass a Func<T> or a "Action".

Thanks anyway!

br--scamb

---

br--scamb