diff --git a/EnumerableAsyncProcessor.UnitTests/AsyncEnumerableProcessorTests.cs b/EnumerableAsyncProcessor.UnitTests/AsyncEnumerableProcessorTests.cs index 30db4f6..c11940a 100644 --- a/EnumerableAsyncProcessor.UnitTests/AsyncEnumerableProcessorTests.cs +++ b/EnumerableAsyncProcessor.UnitTests/AsyncEnumerableProcessorTests.cs @@ -104,6 +104,71 @@ public async Task SelectAsync_ProcessInParallel_ReturnsAllTransformedItems() await Assert.That(results.OrderBy(x => x)).IsEquivalentTo(Enumerable.Range(1, 10).Select(x => x * 2)); } + [Test] + public async Task SelectAsync_BoundedParallelism_PreservesInputOrder() + { + var results = await GenerateAsyncEnumerable(20) + .SelectAsync(async item => + { + await Task.Delay((21 - item) * 2); + return item; + }) + .ProcessInParallel(4) + .ExecuteAsync() + .ToListAsync(); + + await Assert.That(results.SequenceEqual(Enumerable.Range(1, 20))).IsTrue(); + } + + [Test, Timeout(10_000)] + public async Task BoundedParallelism_AppliesBackpressureToSource(CancellationToken cancellationToken) + { + const int maxConcurrency = 2; + + var producedCount = 0; + var startedCount = 0; + var workersStarted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var releaseWorkers = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + + async IAsyncEnumerable Source([EnumeratorCancellation] CancellationToken token = default) + { + for (var i = 0; i < 100; i++) + { + token.ThrowIfCancellationRequested(); + Interlocked.Increment(ref producedCount); + yield return i; + await Task.Yield(); + } + } + + var processingTask = Source(cancellationToken) + .ForEachAsync(async _ => + { + if (Interlocked.Increment(ref startedCount) == maxConcurrency) + { + workersStarted.TrySetResult(); + } + + await releaseWorkers.Task; + }, cancellationToken) + .ProcessInParallel(maxConcurrency) + .ExecuteAsync(); + + try + { + await workersStarted.Task.WaitAsync(TimeSpan.FromSeconds(3), cancellationToken); + await Task.Delay(100, cancellationToken); + + await Assert.That(producedCount).IsLessThanOrEqualTo((maxConcurrency * 2) + 1); + } + finally + { + releaseWorkers.TrySetResult(); + } + + await processingTask; + } + [Test] public async Task ForEachAsync_ProcessInParallel_WithHighConcurrency_HandlesCorrectly() { @@ -213,6 +278,34 @@ public async Task ForEachAsync_WithException_PropagatesException() var exception = await Assert.ThrowsAsync(async () => await task); await Assert.That(exception!.Message).IsEqualTo("Test exception"); } + + [Test, Timeout(10_000)] + public async Task ForEachAsync_WithConcurrentExceptions_PropagatesOriginalException( + CancellationToken cancellationToken) + { + var startedCount = 0; + var workersStarted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var releaseWorkers = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + + var task = GenerateAsyncEnumerable(2) + .ForEachAsync(async _ => + { + if (Interlocked.Increment(ref startedCount) == 2) + { + workersStarted.TrySetResult(); + } + + await releaseWorkers.Task.WaitAsync(cancellationToken); + throw new InvalidOperationException("Concurrent failure"); + }) + .ProcessInParallel(2) + .ExecuteAsync(); + + await workersStarted.Task.WaitAsync(cancellationToken); + releaseWorkers.TrySetResult(); + + await Assert.ThrowsAsync(() => task); + } [Test] public async Task ForEachAsync_ProcessInParallel_UnboundedConcurrency_ProcessesAllItems() diff --git a/EnumerableAsyncProcessor/AsyncEnumerableWorkerPool.cs b/EnumerableAsyncProcessor/AsyncEnumerableWorkerPool.cs new file mode 100644 index 0000000..6ae9bc2 --- /dev/null +++ b/EnumerableAsyncProcessor/AsyncEnumerableWorkerPool.cs @@ -0,0 +1,229 @@ +using System.Collections.Concurrent; +using System.Runtime.CompilerServices; +using System.Runtime.ExceptionServices; +using System.Threading.Channels; + +namespace EnumerableAsyncProcessor; + +/// +/// Processes asynchronous sources with a bounded channel and a fixed set of workers. +/// Source read-ahead and queued results stay proportional to worker count. +/// +internal static class AsyncEnumerableWorkerPool +{ + internal static async Task ProcessAsync( + IAsyncEnumerable items, + Func taskSelector, + int workerCount, + CancellationToken cancellationToken) + { + using var pipelineCancellation = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); + var pipelineToken = pipelineCancellation.Token; + var channel = CreateChannel(workerCount); + var exceptions = new ConcurrentQueue(); + var wasCanceled = 0; + var workers = StartWorkers(channel.Reader, taskSelector, workerCount, exceptions, () => Interlocked.Exchange(ref wasCanceled, 1), pipelineToken); + + try + { + try + { + await foreach (var item in items.WithCancellation(pipelineToken).ConfigureAwait(false)) + { + await channel.Writer.WriteAsync(item, pipelineToken).ConfigureAwait(false); + } + } + catch (OperationCanceledException) + { + Interlocked.Exchange(ref wasCanceled, 1); + } + catch (Exception exception) + { + exceptions.Enqueue(exception); + } + finally + { + channel.Writer.TryComplete(); + } + + await Task.WhenAll(workers).ConfigureAwait(false); + + ThrowIfFailed(exceptions, wasCanceled, cancellationToken); + } + finally + { + pipelineCancellation.Cancel(); + channel.Writer.TryComplete(); + } + } + + internal static async IAsyncEnumerable ProcessResultsAsync( + IAsyncEnumerable items, + Func> taskSelector, + int workerCount, + [EnumeratorCancellation] CancellationToken cancellationToken) + { + using var pipelineCancellation = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); + var pipelineToken = pipelineCancellation.Token; + var channel = CreateChannel>(workerCount); + var workers = StartResultWorkers(channel.Reader, taskSelector, workerCount, pipelineToken); + var pendingResults = new Queue>(workerCount); + + try + { + await foreach (var item in items.WithCancellation(pipelineToken).ConfigureAwait(false)) + { + var completionSource = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + await channel.Writer.WriteAsync(new ResultWorkItem(item, completionSource), pipelineToken).ConfigureAwait(false); + pendingResults.Enqueue(completionSource.Task); + + if (pendingResults.Count == workerCount) + { + yield return await pendingResults.Dequeue().ConfigureAwait(false); + } + } + + channel.Writer.TryComplete(); + + while (pendingResults.TryDequeue(out var resultTask)) + { + yield return await resultTask.ConfigureAwait(false); + } + + await Task.WhenAll(workers).ConfigureAwait(false); + } + finally + { + pipelineCancellation.Cancel(); + channel.Writer.TryComplete(); + + try + { + await Task.WhenAll(workers).ConfigureAwait(false); + } + catch (OperationCanceledException) when (pipelineToken.IsCancellationRequested) + { + // Expected when enumeration is canceled or the consumer stops early. + } + } + } + + private static Channel CreateChannel(int capacity) + { + return Channel.CreateBounded(new BoundedChannelOptions(capacity) + { + SingleWriter = true, + SingleReader = false, + FullMode = BoundedChannelFullMode.Wait, + AllowSynchronousContinuations = false + }); + } + + private static Task[] StartWorkers( + ChannelReader reader, + Func taskSelector, + int workerCount, + ConcurrentQueue exceptions, + Action recordCancellation, + CancellationToken cancellationToken) + { + var workers = new Task[workerCount]; + + for (var i = 0; i < workerCount; i++) + { + workers[i] = Task.Run(async () => + { + await foreach (var item in reader.ReadAllAsync(cancellationToken).ConfigureAwait(false)) + { + Task? task = null; + + try + { + task = taskSelector(item); + await task.ConfigureAwait(false); + } + catch (OperationCanceledException) + { + recordCancellation(); + } + catch (Exception exception) + { + EnqueueExceptions(exceptions, task, exception); + } + } + }, cancellationToken); + } + + return workers; + } + + private static Task[] StartResultWorkers( + ChannelReader> reader, + Func> taskSelector, + int workerCount, + CancellationToken cancellationToken) + { + var workers = new Task[workerCount]; + + for (var i = 0; i < workerCount; i++) + { + workers[i] = Task.Run(async () => + { + await foreach (var workItem in reader.ReadAllAsync(cancellationToken).ConfigureAwait(false)) + { + Task? task = null; + + try + { + task = taskSelector(workItem.Input); + workItem.CompletionSource.TrySetResult(await task.ConfigureAwait(false)); + } + catch (Exception exception) + { + workItem.CompletionSource.TrySetFromFault(task, exception, cancellationToken); + } + } + }, cancellationToken); + } + + return workers; + } + + private static void EnqueueExceptions( + ConcurrentQueue exceptions, + Task? task, + Exception exception) + { + if (task is { IsFaulted: true }) + { + foreach (var innerException in task.Exception!.InnerExceptions) + { + exceptions.Enqueue(innerException); + } + + return; + } + + exceptions.Enqueue(exception); + } + + private static void ThrowIfFailed( + ConcurrentQueue exceptions, + int wasCanceled, + CancellationToken cancellationToken) + { + if (exceptions.TryDequeue(out var firstException)) + { + ExceptionDispatchInfo.Capture(firstException).Throw(); + } + + if (wasCanceled != 0) + { + throw new OperationCanceledException(cancellationToken); + } + } + + private readonly record struct ResultWorkItem( + TInput Input, + TaskCompletionSource CompletionSource); +} diff --git a/EnumerableAsyncProcessor/Extensions/AsyncEnumerableExtensions.cs b/EnumerableAsyncProcessor/Extensions/AsyncEnumerableExtensions.cs index 37a6698..0bf1eea 100644 --- a/EnumerableAsyncProcessor/Extensions/AsyncEnumerableExtensions.cs +++ b/EnumerableAsyncProcessor/Extensions/AsyncEnumerableExtensions.cs @@ -181,68 +181,12 @@ public static async Task> ProcessInParallel( CancellationToken cancellationToken = default) { var results = new List(); - - if (maxConcurrency.HasValue) - { - // Rate-limited parallel processing - using var semaphore = new SemaphoreSlim(maxConcurrency.Value, maxConcurrency.Value); - 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(() => - { - try - { - return capturedItem; - } - finally - { - semaphore.Release(); - } - }, cancellationToken); - - tasks.Add(task); - } - - if (tasks.Count > 0) - { - var taskResults = await Task.WhenAll(tasks).ConfigureAwait(false); - results.AddRange(taskResults); - } - } - else + await foreach (var item in items.WithCancellation(cancellationToken).ConfigureAwait(false)) { - // Unbounded parallel processing - var tasks = new List>(); - - await foreach (var item in items.WithCancellation(cancellationToken).ConfigureAwait(false)) - { - var capturedItem = item; - - Task task; - if (scheduleOnThreadPool) - { - task = Task.Run(() => Task.FromResult(capturedItem), cancellationToken); - } - else - { - task = Task.FromResult(capturedItem); - } - - tasks.Add(task); - } - - if (tasks.Count > 0) - { - var taskResults = await Task.WhenAll(tasks).ConfigureAwait(false); - results.AddRange(taskResults); - } + results.Add(item); } - + return results; } @@ -283,34 +227,13 @@ public static async Task> ProcessInParallel( if (maxConcurrency.HasValue) { - // Rate-limited parallel processing - using var semaphore = new SemaphoreSlim(maxConcurrency.Value, maxConcurrency.Value); - 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 () => - { - try - { - return await taskSelector(capturedItem).ConfigureAwait(false); - } - finally - { - semaphore.Release(); - } - }, cancellationToken); - - tasks.Add(task); - } - - if (tasks.Count > 0) + await foreach (var result in AsyncEnumerableWorkerPool.ProcessResultsAsync( + items, + taskSelector, + maxConcurrency.Value, + cancellationToken).ConfigureAwait(false)) { - var taskResults = await Task.WhenAll(tasks).ConfigureAwait(false); - results.AddRange(taskResults); + results.Add(result); } } else @@ -325,7 +248,7 @@ public static async Task> ProcessInParallel( Task task; if (scheduleOnThreadPool) { - task = Task.Run(async () => await taskSelector(capturedItem).ConfigureAwait(false), cancellationToken); + task = Task.Run(() => taskSelector(capturedItem), cancellationToken); } else { diff --git a/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/AsyncEnumerableParallelProcessor.cs b/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/AsyncEnumerableParallelProcessor.cs index ec1a710..5e395be 100644 --- a/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/AsyncEnumerableParallelProcessor.cs +++ b/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/AsyncEnumerableParallelProcessor.cs @@ -30,40 +30,13 @@ public async Task ExecuteAsync() if (_maxConcurrency.HasValue) { - // Rate-limited parallel processing - using var semaphore = new SemaphoreSlim(_maxConcurrency.Value, _maxConcurrency.Value); - var tasks = new List(); + await AsyncEnumerableWorkerPool.ProcessAsync( + _items, + _taskSelector, + _maxConcurrency.Value, + cancellationToken).ConfigureAwait(false); - 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); - } - } + return; } else { diff --git a/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/ResultProcessors/ResultAsyncEnumerableParallelProcessor.cs b/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/ResultProcessors/ResultAsyncEnumerableParallelProcessor.cs index f3a20aa..0bbba83 100644 --- a/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/ResultProcessors/ResultAsyncEnumerableParallelProcessor.cs +++ b/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/ResultProcessors/ResultAsyncEnumerableParallelProcessor.cs @@ -1,4 +1,3 @@ -using System.Collections.Concurrent; using EnumerableAsyncProcessor.Extensions; namespace EnumerableAsyncProcessor.RunnableProcessors.AsyncEnumerable.ResultProcessors; @@ -28,106 +27,59 @@ internal ResultAsyncEnumerableParallelProcessor( public async IAsyncEnumerable ExecuteAsync() { var cancellationToken = _cancellationTokenSource.Token; - var tasks = new List>(); - if (_maxConcurrency.HasValue) { - // Rate-limited parallel processing - using var semaphore = new SemaphoreSlim(_maxConcurrency.Value, _maxConcurrency.Value); + await foreach (var result in AsyncEnumerableWorkerPool.ProcessResultsAsync( + _items, + _taskSelector, + _maxConcurrency.Value, + cancellationToken).ConfigureAwait(false)) + { + yield return result; + } + + yield break; + } - try + var tasks = new List>(); + + // Unbounded parallel processing + try + { + await foreach (var item in _items.WithCancellation(cancellationToken).ConfigureAwait(false)) { - await foreach (var item in _items.WithCancellation(cancellationToken).ConfigureAwait(false)) + var capturedItem = item; + + Task task; + if (_scheduleOnThreadPool) { - await semaphore.WaitAsync(cancellationToken).ConfigureAwait(false); - - var capturedItem = item; - // Use Task.Run to ensure parallelism and prevent blocking - var task = Task.Run(async () => - { - try - { - return await _taskSelector(capturedItem).ConfigureAwait(false); - } - finally - { - semaphore.Release(); - } - }, cancellationToken); - tasks.Add(task); - - // Yield completed results - while (tasks.Count > 0 && tasks[0].IsCompleted) - { - var completedTask = tasks[0]; - tasks.RemoveAt(0); - yield return await completedTask.ConfigureAwait(false); - } + task = Task.Run(() => _taskSelector(capturedItem), cancellationToken); } - - // Yield remaining results - foreach (var task in tasks) + else { - yield return await task.ConfigureAwait(false); + task = _taskSelector(capturedItem); } + + tasks.Add(task); } - finally + + // Yield all results as they complete + await foreach (var result in tasks.ToIAsyncEnumerable(cancellationToken).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 - } - } + yield return result; } } - else + finally { - // Unbounded parallel processing - try + if (tasks.Count > 0) { - await foreach (var item in _items.WithCancellation(cancellationToken).ConfigureAwait(false)) + try { - var capturedItem = item; - - Task task; - if (_scheduleOnThreadPool) - { - task = Task.Run(() => _taskSelector(capturedItem), cancellationToken); - } - else - { - task = _taskSelector(capturedItem); - } - - tasks.Add(task); + await Task.WhenAll(tasks).ConfigureAwait(false); } - - // Yield all results as they complete - await foreach (var result in tasks.ToIAsyncEnumerable(cancellationToken).ConfigureAwait(false)) - { - yield return result; - } - } - finally - { - if (tasks.Count > 0) + catch { - try - { - await Task.WhenAll(tasks).ConfigureAwait(false); - } - catch - { - // Preserve the exception already propagating from enumeration or result consumption. - } + // Preserve the exception already propagating from enumeration or result consumption. } } }