Calling many web services asynchronously

Question

I need to process a bunch of completed orders (thousands). To make it faster, I'm using a SemaphoreSlim to schedule a number of orders to be processed in parallel. Every order goes through multiple web service calls to various APIs (SOAP, REST). Then I need to take the returned data and process it into a PDF. Here is what I am doing currently (note that this is in an ASP.NET app):

private async Task ProcessOrders(List<Orders> orders)
{
    var allTasks = new List<Task>();
    var throttler = new SemaphoreSlim(initialCount: 5);

    foreach (var order in orders.Where(x => !x.HasError))
    {
        // do an async wait until we can schedule again
        await throttler.WaitAsync();

        // using Task.Run() to run the lambda in its own parallel
        // flow on the thread pool
        allTasks.Add(Task.Run(async () =>
        {
            try
            {
                // Call external web service here
                // Build order report PDF here
            }
            finally
            {
                throttler.Release();
            }
        }));
    }

    await Task.WhenAll(allTasks);
}

However, I'm new to TPL and I don't know if this is the right way to do something like this. If it's relevant, this app is hosted on Amazon's AWS cloud.

Answer 1

I'm not all that familiar with async/await (I really need to get going on this), but I find your code pretty clear, except why the magic number 5 is being used for an initialCount.

I like your usage of var, but the comments are a little redundant, they say nothing much that the code doesn't already tell.

It would have been nice to see the actual code in place of the placeholder comments in the try block - I'd expect to see a method call here; if that comment stands for code that actually implements what's described, you need to factor it out.

Also I would have indented the async task's body, but that's more of a nitpick.

I would probably also have introduced a var validOrders = orders.Where(order => !order.HasError); to make the foreach loop clearer. Not sure I like x for a name in the lambda: try to use meaningful names everywhere.

Answer 2

SemaphoreSlim implements IDisposable and should be disposed of, preferably using the using statement.

Answer 3

The fact that the lambda you're passing to Task.Run is async, along with the comment, implies that the work being done is not CPU bound work, but rather is IO bound work. Given this, there is no need to call Task.Run here.

Task.Run allows you to turn synchronous CPU bound work into asynchronous CPU bound work. However, the work that you have is neither synchronous nor is it CPU bound.

The change is as simple as removing the call to Task.Run. This will remove the unneeded overhead of scheduling the IO operation on a thread pool thread without any loss of functionality.

---

An unrelated refactor that you could make is to use LINQ to transform your sequence of items into a sequence of tasks representing the completion of some work on those items, rather than creating a list, iterating the items, and adding the appropriate Task to the list.

Combining both refactors this gives us:

private async Task ProcessOrders(List<Orders> orders)
{
    var throttler = new SemaphoreSlim(initialCount: 5);

    var tasks = orders.Where(order => !order.HasError)
        .Select(async order =>
        {
            await throttler.WaitAsync();
            try
            {
                await ProcessOrder(order);
            }
            finally
            {
                throttler.Release();
            }
        });

    await Task.WhenAll(tasks);
}

Answer 4

I think that "asynchronous foreach with limited degree of parallelism" is general enough for a helper method. With that, your code could look something like:

private async Task ProcessOrders(List<Orders> orders)
{
    await AsyncParallel.ForEach(
        orders.Where(x => !x.HasError), 5,
        async order =>
        {
            // Call external web service here
            // Build order report PDF here
        });
}

Answer 5

You can also use the task factory for this, as shown below.

private void ProcessOrders(List<Orders> orders)
{
    var allTasks = new List<Task>();
    foreach (var order in orders.Where(x => !x.HasError))
    {
        allTasks.Add(Task.Factory.StartNew(() =>
           // process order    
        ));
    }

    Task.WaitAll(allTasks.ToArray());
}