Parallel.ForEachAsync with BlockingCollection

[woutervanranst]

What is the best practice for the new Parallel.ForEachAsync in combination with a BlockingCollection?

I'm looking to update my implementation based on @stephentoub blog https://devblogs.microsoft.com/pfxteam/implementing-a-simple-foreachasync-part-2/ with the new net6 implementation but not sure whether I can pass the BlockingCollection as the source or should still use the GetConsumingEnumerable()

[stephentoub]

but not sure whether I can pass the BlockingCollection as the source or should still use the GetConsumingEnumerable()

Just passing the collection will enumerate but not consume the contents as it's doing so.

BlockingCollection

You'd be better off using a channel rather than having threads block with a "blocking" collection waiting for the next item to arrive. The primary purpose of ForEachAsync is to enable iteration without blocking.

[woutervanranst]

Stephen thanks for the reply. My initial ask was ill formulated. In any case I'll be looking into leveraging Channels.

To clarify -- GetConsumingEnumerable() is indeed the functionality that we need, but you warn of performance hits in this post:

First, BlockingCollection’s GetConsumingEnumerable implementation is using BlockingCollection’s internal synchronization which already supports multiple consumers concurrently, but ForEach doesn’t know that, and its enumerable-partitioning logic also needs to take a lock while accessing the enumerable. As such, there’s more synchronization here than is actually necessary, resulting in a potentially non-negligable performance hit. Second, the partitioning algorithm employed by default by both Parallel.ForEach and PLINQ use chunking in order to minimize synchronization costs: rather than taking the lock once per element, it’ll take the lock, grab a group of elements (a chunk), and then release the lock. While this design can help with overall throughput, for scenarios that are focused more on low latency, that chunking can be prohibitive.

In the current codebase, we followed your advice and went with the Parallel.ForEach(GetConsumingPartitioner<T>) overload. net6 Parallel.ForEachAsync does not provide an overload for a Partitioner, so this leaves me wondering:

• is it now 'ok' to use a ConsumingEnumerable() (with or without performance hit as net6 somehow factors in the interlocking);
• is it now 'ok' to use GetConsumingPartitioner().GetDynamicPartitions() instead, which is an IEnumerable<T> (seems clunky);
• it is preferable to stay with the Parallel.ForEach() and the Partitioner workaround for now until (maybe) a Partitioner overload becomes available in a future version

[stephentoub]

Nothing has changed in this regard for Parallel.ForEach and GetConsumingEnumerable.

[woutervanranst]

thanks Stephen