Concurrent download/processing in C#

Question

I'm looking for the fastest and most reliable approach to downloading 1000 remote webpages (using HttpWebRequest) concurrently using C#, writing them to individual local files and running some processing code once all files have been downloaded, while making the best use of parallelism and non-blocking concurrency available.

The server is a quad core (vCPU) VPS running Windows 2008 and .NET 4.0 (can't use the newer async/await stuff).

What do you suggest?

Update: Options proposed so far are: Reactive Extensions (Rx), Async CTP, TPL.

Looks like Async CTP would be the ideal way to do it, followed by Rx and TPL. What say guys?

Answer 1

No matter which async approach you end up using, don't forget that you need to increase the max connections allowed as the default is 2 per domain. So if you make a lot of calls against a single domain, you will be rate limited to that.

You can fix this in a standalone (non-ASP.NET) app using basic config:

<system.net>
   <connectionManagement>
       <add address="*" maxconnections="200" />
   </connectionManagement>
</system.net>

However, if you're in ASP.NET this will not work as expected since the default <processModel autoConfig="true" ...> attribute will cause it to auto configure to 12 per core which, while better than 2 total, still might not suit your needs. So then you will have to use the code-based approach in something like your Application_Start:

ServicePointManager.DefaultConnectionLimit = 200;

NOTE: this code based approach also works equally well for non-ASP.NET apps, so you could use it as a "universal" solution if you want to avoid .config.

Answer 2

I would use Rx for that task.

string[] webpages = { "http://www.google.com", "http://www.spiegel.de"};

webpages
    .Select(w => FetchWebPage(w))
    .ForkJoin()
    .Subscribe(x => /*This runs when all webpages have been fetched*/  Console.WriteLine(x));

Or if you like to control the concurrency to process max 4 requests concurrently as svick suggested you could change it to this:

Observable.ForkJoin(
    webpages
        .Select(w => FetchWebPage(w))
        .Merge(4))
        .Subscribe(x => /*This runs when all webpages have been fetched*/  Console.WriteLine(x));   

You also neeed a helper method to transform from the regular async way to the Rx way

public static IObservable<string> FetchWebPage(string address)
{
    var client = new WebClient();

    return Observable.Create<string>(observer =>
    {
        DownloadStringCompletedEventHandler handler = (sender, args) =>
        {
            if (args.Cancelled)
                observer.OnCompleted();
            else if(args.Error != null)
                observer.OnError(args.Error);
            else
            {
                observer.OnNext(args.Result);
                observer.OnCompleted();
            }
        };

        client.DownloadStringCompleted += handler;

        try
        {
            client.DownloadStringAsync(new Uri(address));
        }
        catch (Exception ex)
        {
            observer.OnError(ex);
        }

        return () => client.DownloadStringCompleted -= handler;
    });
}