How to concurrently complete HTTP calls on an observable collection?

Question

In the WPF .net core app there is the following:

An Observable Collection of items (itemObservCollection).
A static readonly HttpClient _httpclient
XML Responses

I am making a URL call to the api on each item in the observable collection (0 to 1000 items in collection). The return is XML. The XML is parsed using XElement. The property values in the observable collection are updated from the XML.

Task.Run is used to run the operation off the UI thread. Parallel.Foreach is used to make the calls in Parallel.

I feel I have made the solution overly complicated. Is there a way to simplify this? UpdateItems() is called from a button click.

private async Task UpdateItems()
{
    try
    {
        await Task.Run(() => Parallel.ForEach(itemObservCollection, new ParallelOptions { MaxDegreeOfParallelism = 12 }, async item =>
        {
            try
            {
                var apiRequestString = $"http://localhost:6060/" + item.Name;
                HttpResponseMessage httpResponseMessage = await _httpclient.GetAsync(apiRequestString);
                var httpResponseStream = await httpResponseMessage.Content.ReadAsStreamAsync();
                StringBuilder sb = new StringBuilder(1024);
                XElement doc = XElement.Load(httpResponseStream);
                foreach (var elem in doc.Descendants())
                {
                    if (elem.Name == "ItemDetails")
                    {
                        var itemUpdate = itemObservCollection.FirstOrDefault(updateItem => updateItem.Name == item.Name);
                        if (itemUpdate != null)
                        {
                            itemUpdate.Price = decimal.Parse(elem.Attribute("Price").Value);
                            itemUpdate.Quantity = int.Parse(elem.Attribute("Quantity").Value);
                        }
                    }
                }
            }
            catch (Exception ex)
            {
                LoggerTextBlock.Text = ('\n' + ex.ToString());
            }
        }));
    }
    catch (Exception ex)
    {
        LoggerTextBlock.Text = ('\n' + ex.ToString());
    }
}

It's generally a bad idea to mix Task.Run with Parallel.ForEach. — Fildor, Sep 24 '20 at 11:08
`$"http://localhost:6060/" + item.Name;` ?? Why not `$"http://localhost:6060/{item.Name}";` ? — Fildor, Sep 24 '20 at 11:13
Using `Parallel.ForEach` with async operations is wrong and actually a bug. `Parallel.ForEach` is meant for *data parallelism* : processing a lot of data using all cores at the same time. It doesn't block, it uses the current thread to process the data. It works by partitioning the data into roughly as many partitions as there are data and feeding each partition to a separate worker thread. Using it with IO operations just blocks the workers without doing anything — Panagiotis Kanavos, Sep 24 '20 at 11:23
The *serious* bug is that `Parallel.ForEach` doesn't handle asynchronous operations, because it wouldn't make any sense to do so. The lambdas you pass to it are essentially `async void` methods that can't be awaited. This code fires off all almost at the same time and never awaits for them to finish — Panagiotis Kanavos, Sep 24 '20 at 11:25
What are you trying to do? Throttle multiple HTTP requests? You could do that with eg an ActionBlock with a `MaxDegreeOfParallelism` option set to the number of concurrent operations you want. There shouldn't be any need to use the collection inside the async method — Panagiotis Kanavos, Sep 24 '20 at 11:27
`Task.Run is used to run the operation off the UI thread.` no, Task.Run is used to run on a *background* thread — Panagiotis Kanavos, Sep 24 '20 at 11:31
This is most definitely *not* an opinion-based question. The current code is wrong — Panagiotis Kanavos, Sep 24 '20 at 11:31
@Fildor the answer is simply "yes", that's not an opinion. The problem is just the title. In fact, I suspect there are several duplicate questions that answer the real problem — Panagiotis Kanavos, Sep 24 '20 at 11:35
Take a look at this: [How to limit the amount of concurrent async I/O operations?](https://stackoverflow.com/questions/10806951/how-to-limit-the-amount-of-concurrent-async-i-o-operations). — Theodor Zoulias, Sep 24 '20 at 12:02
The `SemaphoreSlim` is sufficient for throttling asynchronous operations, but if you are searching for something more powerful keep in mind the TPL Dataflow library. [`Here`](https://stackoverflow.com/questions/60929044/c-sharp-parallel-foreach-memory-usage-keeps-growing/60930992#60930992) is an example of using this library for downloading stuff from the web. — Theodor Zoulias, Sep 24 '20 at 12:11
@Panagiotis Kanavos.I am trying to update the Observable Collection bound to the UI, using new information returned from HttpRequests. I would like to accomplish this as quick as possible, while not blocking the UI thread. — adventuresncode, Sep 24 '20 at 13:01
The consensus seems to be that Task.Run and Parallel.ForEach are incorrect. I will look at TPL Dataflow. Thanks @TheodorZoulias, Fildor, and Panagiotis Kanavos — adventuresncode, Sep 24 '20 at 13:04
@adventuresncode you can't do that with Parallel.ForEach, that's not what it's for. Use an ActionBlock with a DOP of 3. — Panagiotis Kanavos, Sep 24 '20 at 13:04
A mistake - the DOP parameter isn't visible. I probably mixed up the value with another question. The DOP could be anything the network or remote servers allow — Panagiotis Kanavos, Sep 24 '20 at 13:15
This is a call to the localhost so it looks to be all processed on the one pc. There should be no network latency to speak of. Have you tried it with just one at a time in a loop. Call and await response, update then loop for the next? You don't get multi threading for free and it could be as fast just one at a time. Might even be faster that way. — Andy, Sep 24 '20 at 15:10
@Andy I have tried it one at a time in a loop, however it is much slower. — adventuresncode, Sep 28 '20 at 07:23
While learning how to implement an ActionBlock, @mm8 provided a viable solution with Task.WhenAll. I will try and setup ActionBlocks to complete this task as well. I will update if successful. — adventuresncode, Sep 28 '20 at 07:34

mm8 · Accepted Answer · 2020-09-24T14:33:02.280

2

You could create an array of tasks and await them all using Task.WhenAll.

The following sample code kicks off a task per item in the ObservableCollection<int> and then wait asynchronously for all tasks to finish:

ObservableCollection<int> itemObservCollection = 
    new ObservableCollection<int>(Enumerable.Range(1, 10));

async Task SendAsync()
{
    //query the HTTP API here...
    await Task.Delay(1000);
}

await Task.WhenAll(itemObservCollection.Select(x => SendAsync()).ToArray());

If you want to limit the number of concurrent requests, you could either iterate through a subset of the source collecton to send requests in batches or use a SemaphoreSlim to limit the number of actual concurrent requests:

Task[] tasks = new Task[itemObservCollection.Count];
using (SemaphoreSlim semaphoreSlim = new SemaphoreSlim(12))
{
    for (int i = 0; i < itemObservCollection.Count; ++i)
    {
        async Task SendAsync()
        {
            //query the HTTP API here...
            try
            {
                await Task.Delay(5000);
            }
            finally
            {
                semaphoreSlim.Release();
            }
        }

        await semaphoreSlim.WaitAsync();
        tasks[i] = SendAsync();
    }
    await Task.WhenAll(tasks);
}

edited Sep 24 '20 at 14:33

answered Sep 24 '20 at 14:02

mm8

163,881
10
57
88

1

There is no throttling here. – Johnathan Barclay Sep 24 '20 at 14:07
Throttling as in `MaxDegreeOfParallelism`? That's just a matter of iterating through a subset of the source collecton, for example using the `Take` extension method: `itemObservCollection.Take(10)...` `itemObservCollection.Skip(10)` ... – mm8 Sep 24 '20 at 14:09
1

The OP wants to send requests for every item in the collection, but limit this to 12 concurrently. `Take` wouldn't achieve this. – Johnathan Barclay Sep 24 '20 at 14:15
`Take(x).Select(_ => SendAsync())` will start `x` tasks. You can then await them and start the next `x` tasks and so on depending on your requirements. This will limit the concurrent number of requests. – mm8 Sep 24 '20 at 14:17
1

In that case, you would need to wait for all `x` tasks to complete before starting the next batch. A semaphore would be a better solution. – Johnathan Barclay Sep 24 '20 at 14:20
@JohnathanBarclay: I see what you mean. A `SemaphoreSlim` will do the job in this case. I added an example that does not involving creating background tasks using `Task.Run` to my answer. – mm8 Sep 24 '20 at 14:38
await Task.WhenAll solved the problem. Thank you @mm8 – adventuresncode Sep 28 '20 at 07:24

How to concurrently complete HTTP calls on an observable collection?

1 Answers1