From 1967e0b8f84e8ab623481f6cc41626ee3d8a6524 Mon Sep 17 00:00:00 2001 From: Tom Longhurst <30480171+thomhurst@users.noreply.github.com> Date: Mon, 11 Aug 2025 20:39:39 +0100 Subject: [PATCH] feat: Enhance IAsyncEnumerable support with full parallel processing capabilities MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Add unbounded parallel processing support for IAsyncEnumerable - Implement ProcessInParallel() overloads matching IEnumerable API - ProcessInParallel() - unbounded concurrency - ProcessInParallel(int?) - configurable concurrency limit - ProcessInParallel(bool) - thread pool scheduling option - ProcessInParallel(int?, bool) - full control - Add batch processing support with ProcessInBatches(int) - Update AsyncEnumerableParallelProcessor to handle unbounded and limited concurrency - Add AsyncEnumerableBatchProcessor for batch-based processing - Update result processors to support new processing modes - Add comprehensive unit tests for all new features - Maintain backward compatibility with existing API This provides complete feature parity between IAsyncEnumerable and IEnumerable processing, allowing developers to use the same fluent API patterns for both collection types. 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude --- .../AsyncEnumerableProcessorTests.cs | 139 ++++++++++++++++++ ...EnumerableActionAsyncProcessorBuilder_1.cs | 57 ++++++- ...EnumerableActionAsyncProcessorBuilder_2.cs | 57 ++++++- .../AsyncEnumerableBatchProcessor.cs | 54 +++++++ .../AsyncEnumerableParallelProcessor.cs | 79 +++++++--- .../ResultAsyncEnumerableBatchProcessor.cs | 65 ++++++++ .../ResultAsyncEnumerableParallelProcessor.cs | 104 ++++++++----- 7 files changed, 485 insertions(+), 70 deletions(-) create mode 100644 EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/AsyncEnumerableBatchProcessor.cs create mode 100644 EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/ResultProcessors/ResultAsyncEnumerableBatchProcessor.cs diff --git a/EnumerableAsyncProcessor.UnitTests/AsyncEnumerableProcessorTests.cs b/EnumerableAsyncProcessor.UnitTests/AsyncEnumerableProcessorTests.cs index 6071058..ccde121 100644 --- a/EnumerableAsyncProcessor.UnitTests/AsyncEnumerableProcessorTests.cs +++ b/EnumerableAsyncProcessor.UnitTests/AsyncEnumerableProcessorTests.cs @@ -215,6 +215,145 @@ public async Task ForEachAsync_WithException_PropagatesException() var exception = await Assert.ThrowsAsync(async () => await task); await Assert.That(exception!.Message).IsEqualTo("Test exception"); } + + [Test] + public async Task ForEachAsync_ProcessInParallel_UnboundedConcurrency_ProcessesAllItems() + { + var processedItems = new List(); + var asyncEnumerable = GenerateAsyncEnumerable(50); + + await asyncEnumerable + .ForEachAsync(async item => + { + await Task.Delay(5); + lock (processedItems) + { + processedItems.Add(item); + } + }) + .ProcessInParallel() // Unbounded concurrency + .ExecuteAsync(); + + await Assert.That(processedItems.Count).IsEqualTo(50); + await Assert.That(processedItems.OrderBy(x => x)).IsEquivalentTo(Enumerable.Range(1, 50)); + } + + [Test] + public async Task SelectAsync_ProcessInParallel_UnboundedConcurrency_ReturnsAllResults() + { + var asyncEnumerable = GenerateAsyncEnumerable(30); + + var results = await asyncEnumerable + .SelectAsync(async item => + { + await Task.Delay(5); + return item * 2; + }) + .ProcessInParallel() // Unbounded concurrency + .ExecuteAsync() + .ToListAsync(); + + await Assert.That(results.Count).IsEqualTo(30); + await Assert.That(results.OrderBy(x => x)).IsEquivalentTo(Enumerable.Range(1, 30).Select(x => x * 2)); + } + + [Test] + public async Task ForEachAsync_ProcessInParallel_WithThreadPoolScheduling_ProcessesAllItems() + { + var processedItems = new List(); + var asyncEnumerable = GenerateAsyncEnumerable(20); + + await asyncEnumerable + .ForEachAsync(async item => + { + await Task.Delay(5); + lock (processedItems) + { + processedItems.Add(item); + } + }) + .ProcessInParallel(scheduleOnThreadPool: true) + .ExecuteAsync(); + + await Assert.That(processedItems.Count).IsEqualTo(20); + await Assert.That(processedItems.OrderBy(x => x)).IsEquivalentTo(Enumerable.Range(1, 20)); + } + + [Test] + public async Task ForEachAsync_ProcessInBatches_ProcessesAllItemsInBatches() + { + var processedBatches = new List(); + var asyncEnumerable = GenerateAsyncEnumerable(25); + + await asyncEnumerable + .ForEachAsync(async item => + { + await Task.Delay(5); + lock (processedBatches) + { + processedBatches.Add(item); + } + }) + .ProcessInBatches(5) + .ExecuteAsync(); + + await Assert.That(processedBatches.Count).IsEqualTo(25); + await Assert.That(processedBatches.OrderBy(x => x)).IsEquivalentTo(Enumerable.Range(1, 25)); + } + + [Test] + public async Task SelectAsync_ProcessInBatches_ReturnsAllResultsInBatches() + { + var asyncEnumerable = GenerateAsyncEnumerable(23); + + var results = await asyncEnumerable + .SelectAsync(async item => + { + await Task.Delay(5); + return item * 3; + }) + .ProcessInBatches(5) + .ExecuteAsync() + .ToListAsync(); + + await Assert.That(results.Count).IsEqualTo(23); + // Batches maintain order within batch, so results should be in order + await Assert.That(results).IsEquivalentTo(Enumerable.Range(1, 23).Select(x => x * 3)); + } + + [Test] + public async Task ProcessInParallel_NullableConcurrency_WorksCorrectly() + { + var asyncEnumerable = GenerateAsyncEnumerable(15); + var processedCount = 0; + + // Test with null concurrency (unbounded) + await asyncEnumerable + .ForEachAsync(async item => + { + await Task.Delay(5); + Interlocked.Increment(ref processedCount); + }) + .ProcessInParallel((int?)null) + .ExecuteAsync(); + + await Assert.That(processedCount).IsEqualTo(15); + + // Reset and test with specified concurrency + processedCount = 0; + asyncEnumerable = GenerateAsyncEnumerable(15); + + await asyncEnumerable + .ForEachAsync(async item => + { + await Task.Delay(5); + Interlocked.Increment(ref processedCount); + }) + .ProcessInParallel((int?)5) + .ExecuteAsync(); + + await Assert.That(processedCount).IsEqualTo(15); + } } internal static class AsyncEnumerableExtensionsForTests diff --git a/EnumerableAsyncProcessor/Builders/AsyncEnumerableActionAsyncProcessorBuilder_1.cs b/EnumerableAsyncProcessor/Builders/AsyncEnumerableActionAsyncProcessorBuilder_1.cs index c1c0b63..a8c1240 100644 --- a/EnumerableAsyncProcessor/Builders/AsyncEnumerableActionAsyncProcessorBuilder_1.cs +++ b/EnumerableAsyncProcessor/Builders/AsyncEnumerableActionAsyncProcessorBuilder_1.cs @@ -21,20 +21,54 @@ public AsyncEnumerableActionAsyncProcessorBuilder( } /// - /// Process items in parallel with a specified level of parallelism. + /// Process items in parallel without concurrency limits. /// + /// An async processor configured for parallel execution. + public IAsyncEnumerableProcessor ProcessInParallel() + { + return ProcessInParallel(null, false); + } + + /// + /// Process items in parallel without concurrency limits. + /// + /// If true, schedules tasks on thread pool to prevent blocking. Default is false for maximum performance. + /// An async processor configured for parallel execution. + public IAsyncEnumerableProcessor ProcessInParallel(bool scheduleOnThreadPool) + { + return ProcessInParallel(null, scheduleOnThreadPool); + } + + /// + /// Process items in parallel with specified concurrency limit. + /// + /// Maximum concurrent operations. + /// An async processor configured for parallel execution. public IAsyncEnumerableProcessor ProcessInParallel(int maxConcurrency) { - return new AsyncEnumerableParallelProcessor( - _items, _taskSelector, maxConcurrency, _cancellationTokenSource); + return ProcessInParallel((int?)maxConcurrency, false); } /// - /// Process items in parallel with default concurrency (processor count). + /// Process items in parallel with specified concurrency limit. /// - public IAsyncEnumerableProcessor ProcessInParallel() + /// Maximum concurrent operations. + /// An async processor configured for parallel execution. + public IAsyncEnumerableProcessor ProcessInParallel(int? maxConcurrency) + { + return ProcessInParallel(maxConcurrency, false); + } + + /// + /// Process items in parallel with specified concurrency limit. + /// + /// Maximum concurrent operations. + /// If true, schedules tasks on thread pool to prevent blocking. + /// An async processor configured for parallel execution. + public IAsyncEnumerableProcessor ProcessInParallel(int? maxConcurrency, bool scheduleOnThreadPool) { - return ProcessInParallel(Environment.ProcessorCount); + return new AsyncEnumerableParallelProcessor( + _items, _taskSelector, maxConcurrency, scheduleOnThreadPool, _cancellationTokenSource); } @@ -46,6 +80,17 @@ public IAsyncEnumerableProcessor ProcessOneAtATime() return new AsyncEnumerableOneAtATimeProcessor( _items, _taskSelector, _cancellationTokenSource); } + + /// + /// Process items in batches. + /// + /// The size of each batch. + /// An async processor configured for batch execution. + public IAsyncEnumerableProcessor ProcessInBatches(int batchSize) + { + return new AsyncEnumerableBatchProcessor( + _items, _taskSelector, batchSize, _cancellationTokenSource); + } } #endif \ No newline at end of file diff --git a/EnumerableAsyncProcessor/Builders/AsyncEnumerableActionAsyncProcessorBuilder_2.cs b/EnumerableAsyncProcessor/Builders/AsyncEnumerableActionAsyncProcessorBuilder_2.cs index 915af0a..79e80e0 100644 --- a/EnumerableAsyncProcessor/Builders/AsyncEnumerableActionAsyncProcessorBuilder_2.cs +++ b/EnumerableAsyncProcessor/Builders/AsyncEnumerableActionAsyncProcessorBuilder_2.cs @@ -21,20 +21,54 @@ public AsyncEnumerableActionAsyncProcessorBuilder( } /// - /// Process items in parallel with a specified level of parallelism and return results. + /// Process items in parallel without concurrency limits and return results. /// + /// An async processor configured for parallel execution. + public IAsyncEnumerableProcessor ProcessInParallel() + { + return ProcessInParallel(null, false); + } + + /// + /// Process items in parallel without concurrency limits and return results. + /// + /// If true, schedules tasks on thread pool to prevent blocking. Default is false for maximum performance. + /// An async processor configured for parallel execution. + public IAsyncEnumerableProcessor ProcessInParallel(bool scheduleOnThreadPool) + { + return ProcessInParallel(null, scheduleOnThreadPool); + } + + /// + /// Process items in parallel with specified concurrency limit and return results. + /// + /// Maximum concurrent operations. + /// An async processor configured for parallel execution. public IAsyncEnumerableProcessor ProcessInParallel(int maxConcurrency) { - return new ResultAsyncEnumerableParallelProcessor( - _items, _taskSelector, maxConcurrency, _cancellationTokenSource); + return ProcessInParallel((int?)maxConcurrency, false); } /// - /// Process items in parallel with default concurrency and return results. + /// Process items in parallel with specified concurrency limit and return results. /// - public IAsyncEnumerableProcessor ProcessInParallel() + /// Maximum concurrent operations. + /// An async processor configured for parallel execution. + public IAsyncEnumerableProcessor ProcessInParallel(int? maxConcurrency) + { + return ProcessInParallel(maxConcurrency, false); + } + + /// + /// Process items in parallel with specified concurrency limit and return results. + /// + /// Maximum concurrent operations. + /// If true, schedules tasks on thread pool to prevent blocking. + /// An async processor configured for parallel execution. + public IAsyncEnumerableProcessor ProcessInParallel(int? maxConcurrency, bool scheduleOnThreadPool) { - return ProcessInParallel(Environment.ProcessorCount); + return new ResultAsyncEnumerableParallelProcessor( + _items, _taskSelector, maxConcurrency, scheduleOnThreadPool, _cancellationTokenSource); } @@ -46,6 +80,17 @@ public IAsyncEnumerableProcessor ProcessOneAtATime() return new ResultAsyncEnumerableOneAtATimeProcessor( _items, _taskSelector, _cancellationTokenSource); } + + /// + /// Process items in batches and return results. + /// + /// The size of each batch. + /// An async processor configured for batch execution. + public IAsyncEnumerableProcessor ProcessInBatches(int batchSize) + { + return new ResultAsyncEnumerableBatchProcessor( + _items, _taskSelector, batchSize, _cancellationTokenSource); + } } #endif \ No newline at end of file diff --git a/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/AsyncEnumerableBatchProcessor.cs b/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/AsyncEnumerableBatchProcessor.cs new file mode 100644 index 0000000..0fabd98 --- /dev/null +++ b/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/AsyncEnumerableBatchProcessor.cs @@ -0,0 +1,54 @@ +#if NET6_0_OR_GREATER +using EnumerableAsyncProcessor.Extensions; + +namespace EnumerableAsyncProcessor.RunnableProcessors.AsyncEnumerable; + +public class AsyncEnumerableBatchProcessor : IAsyncEnumerableProcessor +{ + private readonly IAsyncEnumerable _items; + private readonly Func _taskSelector; + private readonly int _batchSize; + private readonly CancellationTokenSource _cancellationTokenSource; + + internal AsyncEnumerableBatchProcessor( + IAsyncEnumerable items, + Func taskSelector, + int batchSize, + CancellationTokenSource cancellationTokenSource) + { + _items = items; + _taskSelector = taskSelector; + _batchSize = batchSize; + _cancellationTokenSource = cancellationTokenSource; + } + + public async Task ExecuteAsync() + { + var cancellationToken = _cancellationTokenSource.Token; + var batch = new List(_batchSize); + + await foreach (var item in _items.WithCancellation(cancellationToken).ConfigureAwait(false)) + { + batch.Add(item); + + if (batch.Count >= _batchSize) + { + await ProcessBatch(batch, cancellationToken).ConfigureAwait(false); + batch = new List(_batchSize); + } + } + + // Process any remaining items in the final batch + if (batch.Count > 0) + { + await ProcessBatch(batch, cancellationToken).ConfigureAwait(false); + } + } + + private async Task ProcessBatch(List batch, CancellationToken cancellationToken) + { + var tasks = batch.Select(item => _taskSelector(item)).ToArray(); + await Task.WhenAll(tasks).ConfigureAwait(false); + } +} +#endif \ No newline at end of file diff --git a/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/AsyncEnumerableParallelProcessor.cs b/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/AsyncEnumerableParallelProcessor.cs index fef545a..598955d 100644 --- a/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/AsyncEnumerableParallelProcessor.cs +++ b/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/AsyncEnumerableParallelProcessor.cs @@ -7,54 +7,87 @@ public class AsyncEnumerableParallelProcessor : IAsyncEnumerableProcesso { private readonly IAsyncEnumerable _items; private readonly Func _taskSelector; - private readonly int _maxConcurrency; + private readonly int? _maxConcurrency; + private readonly bool _scheduleOnThreadPool; private readonly CancellationTokenSource _cancellationTokenSource; internal AsyncEnumerableParallelProcessor( IAsyncEnumerable items, Func taskSelector, - int maxConcurrency, + int? maxConcurrency, + bool scheduleOnThreadPool, CancellationTokenSource cancellationTokenSource) { _items = items; _taskSelector = taskSelector; _maxConcurrency = maxConcurrency; + _scheduleOnThreadPool = scheduleOnThreadPool; _cancellationTokenSource = cancellationTokenSource; } public async Task ExecuteAsync() { var cancellationToken = _cancellationTokenSource.Token; - using var semaphore = new SemaphoreSlim(_maxConcurrency, _maxConcurrency); - var tasks = new List(); + + if (_maxConcurrency.HasValue) + { + // Rate-limited parallel processing + using var semaphore = new SemaphoreSlim(_maxConcurrency.Value, _maxConcurrency.Value); + var tasks = new List(); - try + try + { + await foreach (var item in _items.WithCancellation(cancellationToken).ConfigureAwait(false)) + { + await semaphore.WaitAsync(cancellationToken).ConfigureAwait(false); + + var capturedItem = item; + var task = Task.Run(async () => + { + try + { + await _taskSelector(capturedItem).ConfigureAwait(false); + } + finally + { + semaphore.Release(); + } + }, cancellationToken); + + tasks.Add(task); + } + } + finally + { + // Always wait for all tasks to complete before the using block disposes the semaphore + if (tasks.Count > 0) + { + await Task.WhenAll(tasks).ConfigureAwait(false); + } + } + } + else { + // Unbounded parallel processing + var tasks = new List(); + await foreach (var item in _items.WithCancellation(cancellationToken).ConfigureAwait(false)) { - await semaphore.WaitAsync(cancellationToken).ConfigureAwait(false); - var capturedItem = item; - var task = Task.Run(async () => + + Task task; + if (_scheduleOnThreadPool) { - try - { - // Removed Task.Yield - parallelism is now handled at the processor level - await _taskSelector(capturedItem).ConfigureAwait(false); - } - finally - { - semaphore.Release(); - } - }, cancellationToken); + task = Task.Run(async () => await _taskSelector(capturedItem).ConfigureAwait(false), cancellationToken); + } + else + { + task = _taskSelector(capturedItem); + } tasks.Add(task); } - } - finally - { - // Always wait for all tasks to complete before the using block disposes the semaphore - // This ensures the semaphore is not disposed while tasks are still running + if (tasks.Count > 0) { await Task.WhenAll(tasks).ConfigureAwait(false); diff --git a/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/ResultProcessors/ResultAsyncEnumerableBatchProcessor.cs b/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/ResultProcessors/ResultAsyncEnumerableBatchProcessor.cs new file mode 100644 index 0000000..5916650 --- /dev/null +++ b/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/ResultProcessors/ResultAsyncEnumerableBatchProcessor.cs @@ -0,0 +1,65 @@ +#if NET6_0_OR_GREATER +using EnumerableAsyncProcessor.Extensions; + +namespace EnumerableAsyncProcessor.RunnableProcessors.AsyncEnumerable.ResultProcessors; + +public class ResultAsyncEnumerableBatchProcessor : IAsyncEnumerableProcessor +{ + private readonly IAsyncEnumerable _items; + private readonly Func> _taskSelector; + private readonly int _batchSize; + private readonly CancellationTokenSource _cancellationTokenSource; + + internal ResultAsyncEnumerableBatchProcessor( + IAsyncEnumerable items, + Func> taskSelector, + int batchSize, + CancellationTokenSource cancellationTokenSource) + { + _items = items; + _taskSelector = taskSelector; + _batchSize = batchSize; + _cancellationTokenSource = cancellationTokenSource; + } + + public async IAsyncEnumerable ExecuteAsync() + { + var cancellationToken = _cancellationTokenSource.Token; + var batch = new List(_batchSize); + + await foreach (var item in _items.WithCancellation(cancellationToken).ConfigureAwait(false)) + { + batch.Add(item); + + if (batch.Count >= _batchSize) + { + await foreach (var result in ProcessBatch(batch, cancellationToken).ConfigureAwait(false)) + { + yield return result; + } + batch = new List(_batchSize); + } + } + + // Process any remaining items in the final batch + if (batch.Count > 0) + { + await foreach (var result in ProcessBatch(batch, cancellationToken).ConfigureAwait(false)) + { + yield return result; + } + } + } + + private async IAsyncEnumerable ProcessBatch(List batch, [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken) + { + var tasks = batch.Select(item => _taskSelector(item)).ToArray(); + var results = await Task.WhenAll(tasks).ConfigureAwait(false); + + foreach (var result in results) + { + yield return result; + } + } +} +#endif \ No newline at end of file diff --git a/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/ResultProcessors/ResultAsyncEnumerableParallelProcessor.cs b/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/ResultProcessors/ResultAsyncEnumerableParallelProcessor.cs index f0e396f..5954f68 100644 --- a/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/ResultProcessors/ResultAsyncEnumerableParallelProcessor.cs +++ b/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/ResultProcessors/ResultAsyncEnumerableParallelProcessor.cs @@ -8,77 +8,111 @@ public class ResultAsyncEnumerableParallelProcessor : IAsyncEnu { private readonly IAsyncEnumerable _items; private readonly Func> _taskSelector; - private readonly int _maxConcurrency; + private readonly int? _maxConcurrency; + private readonly bool _scheduleOnThreadPool; private readonly CancellationTokenSource _cancellationTokenSource; internal ResultAsyncEnumerableParallelProcessor( IAsyncEnumerable items, Func> taskSelector, - int maxConcurrency, + int? maxConcurrency, + bool scheduleOnThreadPool, CancellationTokenSource cancellationTokenSource) { _items = items; _taskSelector = taskSelector; _maxConcurrency = maxConcurrency; + _scheduleOnThreadPool = scheduleOnThreadPool; _cancellationTokenSource = cancellationTokenSource; } public async IAsyncEnumerable ExecuteAsync() { var cancellationToken = _cancellationTokenSource.Token; - using var semaphore = new SemaphoreSlim(_maxConcurrency, _maxConcurrency); var tasks = new List>(); - try + if (_maxConcurrency.HasValue) { - await foreach (var item in _items.WithCancellation(cancellationToken).ConfigureAwait(false)) + // Rate-limited parallel processing + using var semaphore = new SemaphoreSlim(_maxConcurrency.Value, _maxConcurrency.Value); + + try { - await semaphore.WaitAsync(cancellationToken).ConfigureAwait(false); - - var capturedItem = item; - // Use Task.Run to ensure parallelism and prevent blocking - var task = Task.Run(async () => + await foreach (var item in _items.WithCancellation(cancellationToken).ConfigureAwait(false)) { - try + await semaphore.WaitAsync(cancellationToken).ConfigureAwait(false); + + var capturedItem = item; + // Use Task.Run to ensure parallelism and prevent blocking + var task = Task.Run(async () => { - return await _taskSelector(capturedItem).ConfigureAwait(false); - } - finally + try + { + return await _taskSelector(capturedItem).ConfigureAwait(false); + } + finally + { + semaphore.Release(); + } + }, cancellationToken); + tasks.Add(task); + + // Yield completed results + while (tasks.Count > 0 && tasks[0].IsCompleted) { - semaphore.Release(); + var completedTask = tasks[0]; + tasks.RemoveAt(0); + yield return await completedTask.ConfigureAwait(false); } - }, cancellationToken); - tasks.Add(task); - - // Yield completed results - while (tasks.Count > 0 && tasks[0].IsCompleted) + } + + // Yield remaining results + foreach (var task in tasks) { - var completedTask = tasks[0]; - tasks.RemoveAt(0); - yield return await completedTask.ConfigureAwait(false); + yield return await task.ConfigureAwait(false); } } - - // Yield remaining results - foreach (var task in tasks) + finally { - yield return await task.ConfigureAwait(false); + // Ensure all tasks complete before the using block disposes the semaphore + // This handles cancellation or exception scenarios + if (tasks.Count > 0) + { + try + { + await Task.WhenAll(tasks).ConfigureAwait(false); + } + catch + { + // Ignore exceptions here as they've already been handled + } + } } } - finally + else { - // Ensure all tasks complete before the using block disposes the semaphore - // This handles cancellation or exception scenarios - if (tasks.Count > 0) + // Unbounded parallel processing + await foreach (var item in _items.WithCancellation(cancellationToken).ConfigureAwait(false)) { - try + var capturedItem = item; + + Task task; + if (_scheduleOnThreadPool) { - await Task.WhenAll(tasks).ConfigureAwait(false); + task = Task.Run(async () => await _taskSelector(capturedItem).ConfigureAwait(false), cancellationToken); } - catch + else { - // Ignore exceptions here as they've already been handled + task = _taskSelector(capturedItem); } + + tasks.Add(task); + } + + // Yield all results as they complete + await foreach (var result in tasks.ToIAsyncEnumerable(cancellationToken).ConfigureAwait(false)) + { + yield return result; } } }