perf: worker-pool processing, array hot paths, O(N) streaming

Measured at 100k items with completed-task selectors (pure
overhead, net9):
- unbounded parallel: 61ms -> 16ms, 444 -> 162 B/item
- maxConcurrency 64: 57ms -> 10ms, 431 -> 164 B/item
- rate-limited 64: 163ms -> 9ms, 1062 -> 156 B/item

Changes:
- Rate-limited, timed and throttled parallel processors now
  run on a fixed pool of P worker loops (WorkerPool) instead
  of queueing one task per item: P Task.Run calls and one
  Interlocked increment per item replace N Task.Run tasks,
  N closures and N semaphore waits. The internal ITaskWrapper
  interface lets one generic helper serve all four wrapper
  structs via constrained calls without boxing.
- TaskWrappers fields are typed as arrays so iteration uses
  struct enumerators and LINQ array fast paths instead of
  interface-dispatched enumerators.
- ToIAsyncEnumerable pre-net9 fallback replaced the O(N^2)
  WhenAny loop (rebuilding the task list every completion)
  with completion-order buckets: O(N), verified on net8 -
  100k results stream in 73ms.
- Throttled ProcessInParallel now validates maxConcurrency
  at build time; previously 0 surfaced as a semaphore
  ArgumentOutOfRangeException at process time.

Author: thomhurst

EnumerableAsyncProcessor/Extensions/EnumerableExtensions.cs

@@ -140,38 +140,29 @@ internal static async IAsyncEnumerable<T> ToIAsyncEnumerable<T>(this IEnumerable
             yield return task.Result;
         }
 #else
-        var managedTasksList = tasks.ToList();
+        // Interleaving via completion-order buckets: each task's continuation claims the next
+        // bucket, so streaming N tasks is O(N) rather than the O(N^2) of a WhenAny loop.
+        var inputTasks = tasks.ToList();
 
-        // Create a cancellation task that will complete when cancellation is requested
-        using var cancellationTcs = new CancellationTokenSource();
-        var cancellationTask = Task.Delay(Timeout.Infinite, cancellationTcs.Token);
-        
-        // Register callback to trigger the cancellation task
-        using var registration = cancellationToken.Register(() => cancellationTcs.Cancel());
+        var buckets = new TaskCompletionSource<Task<T>>[inputTasks.Count];
+        for (var i = 0; i < buckets.Length; i++)
+        {
+            buckets[i] = new TaskCompletionSource<Task<T>>(TaskCreationOptions.RunContinuationsAsynchronously);
+        }
 
-        while (managedTasksList.Count != 0)
+        var nextBucketIndex = -1;
+        foreach (var task in inputTasks)
         {
-            // Check for cancellation before each iteration
-            cancellationToken.ThrowIfCancellationRequested();
-            
-            // Include the cancellation task in WhenAny
-            var allTasks = new List<Task>(managedTasksList.Count + 1);
-            allTasks.AddRange(managedTasksList);
-            allTasks.Add(cancellationTask);
-            
-            var finishedTask = await Task.WhenAny(allTasks).ConfigureAwait(false);
-            
-            // If the cancellation task completed, throw cancellation
-            if (finishedTask == cancellationTask)
-            {
-                cancellationToken.ThrowIfCancellationRequested();
-                // This should not happen as cancellation should throw above, but as a safety measure:
-                throw new OperationCanceledException(cancellationToken);
-            }
-            
-            // Remove and yield the completed task
-            var completedTask = (Task<T>)finishedTask;
-            managedTasksList.Remove(completedTask);
+            _ = task.ContinueWith(
+                completedTask => buckets[Interlocked.Increment(ref nextBucketIndex)].TrySetResult(completedTask),
+                CancellationToken.None,
+                TaskContinuationOptions.ExecuteSynchronously,
+                TaskScheduler.Default);
+        }
+
+        foreach (var bucket in buckets)
+        {
+            var completedTask = await bucket.Task.WaitAsync(cancellationToken).ConfigureAwait(false);
             yield return await completedTask.ConfigureAwait(false);
         }
 #endif

EnumerableAsyncProcessor/RunnableProcessors/Abstract/AbstractAsyncProcessor.cs

@@ -4,7 +4,7 @@ namespace EnumerableAsyncProcessor.RunnableProcessors.Abstract;
 
 public abstract class AbstractAsyncProcessor : AbstractAsyncProcessorBase
 {
-    protected readonly IReadOnlyList<ActionTaskWrapper> TaskWrappers;
+    protected readonly ActionTaskWrapper[] TaskWrappers;
 
     private readonly TaskCompletionSource[] _taskCompletionSources;
 

EnumerableAsyncProcessor/RunnableProcessors/Abstract/AbstractAsyncProcessor_1.cs

@@ -4,7 +4,7 @@ namespace EnumerableAsyncProcessor.RunnableProcessors.Abstract;
 
 public abstract class AbstractAsyncProcessor<TInput> : AbstractAsyncProcessorBase
 {
-    protected readonly IReadOnlyList<ItemTaskWrapper<TInput>> TaskWrappers;
+    protected readonly ItemTaskWrapper<TInput>[] TaskWrappers;
 
     private readonly TaskCompletionSource[] _taskCompletionSources;
 

EnumerableAsyncProcessor/RunnableProcessors/ParallelAsyncProcessor.cs

@@ -1,14 +1,20 @@
 using EnumerableAsyncProcessor.RunnableProcessors.Abstract;
+using EnumerableAsyncProcessor.Validation;
 
 namespace EnumerableAsyncProcessor.RunnableProcessors;
 
 public class ParallelAsyncProcessor : AbstractAsyncProcessor
 {
     private readonly int? _maxConcurrency;
     private readonly bool _scheduleOnThreadPool;
-    
+
     internal ParallelAsyncProcessor(int count, Func<Task> taskSelector, CancellationTokenSource cancellationTokenSource, int? maxConcurrency = null, bool scheduleOnThreadPool = false) : base(count, taskSelector, cancellationTokenSource)
     {
+        if (maxConcurrency is { } concurrencyLimit)
+        {
+            ValidationHelper.ThrowIfNegativeOrZero(concurrencyLimit, nameof(maxConcurrency));
+        }
+
         _maxConcurrency = maxConcurrency;
         _scheduleOnThreadPool = scheduleOnThreadPool;
     }
@@ -38,38 +44,8 @@ await Task.WhenAll(TaskWrappers.Select(taskWrapper =>
             return;
         }
 
-        // Use semaphore for concurrency throttling
-        using var semaphore = new SemaphoreSlim(_maxConcurrency.Value, _maxConcurrency.Value);
-        
-        // Materialize tasks immediately to ensure they all start in parallel (up to concurrency limit)
-        var tasks = _scheduleOnThreadPool
-            ? // Use Task.Run to prevent synchronous code from blocking thread pool threads
-              TaskWrappers.Select(taskWrapper => Task.Run(async () =>
-        {
-            await semaphore.WaitAsync(CancellationToken).ConfigureAwait(false);
-            try
-            {
-                await taskWrapper.Process(CancellationToken).ConfigureAwait(false);
-            }
-            finally
-            {
-                semaphore.Release();
-            }
-              }, CancellationToken)).ToList()
-            : // Direct execution for maximum performance
-              TaskWrappers.Select(async taskWrapper =>
-              {
-                  await semaphore.WaitAsync(CancellationToken).ConfigureAwait(false);
-                  try
-                  {
-                      await taskWrapper.Process(CancellationToken).ConfigureAwait(false);
-                  }
-                  finally
-                  {
-                      semaphore.Release();
-                  }
-              }).ToList(); // Force immediate task creation
-
-        await Task.WhenAll(tasks).ConfigureAwait(false);
+        // Throttled processing runs on a fixed worker pool: P worker tasks instead of
+        // one queued task, closure and semaphore wait per item
+        await WorkerPool.ProcessAsync(TaskWrappers, _maxConcurrency.Value, minimumIterationTime: null, CancellationToken).ConfigureAwait(false);
     }
 }

EnumerableAsyncProcessor/RunnableProcessors/ParallelAsyncProcessor_1.cs

@@ -1,14 +1,20 @@
 using EnumerableAsyncProcessor.RunnableProcessors.Abstract;
+using EnumerableAsyncProcessor.Validation;
 
 namespace EnumerableAsyncProcessor.RunnableProcessors;
 
 public class ParallelAsyncProcessor<TInput> : AbstractAsyncProcessor<TInput>
 {
     private readonly int? _maxConcurrency;
     private readonly bool _scheduleOnThreadPool;
-    
+
     internal ParallelAsyncProcessor(IEnumerable<TInput> items, Func<TInput, Task> taskSelector, CancellationTokenSource cancellationTokenSource, int? maxConcurrency = null, bool scheduleOnThreadPool = false) : base(items, taskSelector, cancellationTokenSource)
     {
+        if (maxConcurrency is { } concurrencyLimit)
+        {
+            ValidationHelper.ThrowIfNegativeOrZero(concurrencyLimit, nameof(maxConcurrency));
+        }
+
         _maxConcurrency = maxConcurrency;
         _scheduleOnThreadPool = scheduleOnThreadPool;
     }
@@ -38,38 +44,8 @@ await Task.WhenAll(TaskWrappers.Select(taskWrapper =>
             return;
         }
 
-        // Use semaphore for concurrency throttling
-        using var semaphore = new SemaphoreSlim(_maxConcurrency.Value, _maxConcurrency.Value);
-        
-        // Materialize tasks immediately to ensure they all start in parallel (up to concurrency limit)
-        var tasks = _scheduleOnThreadPool
-            ? // Use Task.Run to prevent synchronous code from blocking thread pool threads
-              TaskWrappers.Select(taskWrapper => Task.Run(async () =>
-        {
-            await semaphore.WaitAsync(CancellationToken).ConfigureAwait(false);
-            try
-            {
-                await taskWrapper.Process(CancellationToken).ConfigureAwait(false);
-            }
-            finally
-            {
-                semaphore.Release();
-            }
-              }, CancellationToken)).ToList()
-            : // Direct execution for maximum performance
-              TaskWrappers.Select(async taskWrapper =>
-              {
-                  await semaphore.WaitAsync(CancellationToken).ConfigureAwait(false);
-                  try
-                  {
-                      await taskWrapper.Process(CancellationToken).ConfigureAwait(false);
-                  }
-                  finally
-                  {
-                      semaphore.Release();
-                  }
-              }).ToList(); // Force immediate task creation
-
-        await Task.WhenAll(tasks).ConfigureAwait(false);
+        // Throttled processing runs on a fixed worker pool: P worker tasks instead of
+        // one queued task, closure and semaphore wait per item
+        await WorkerPool.ProcessAsync(TaskWrappers, _maxConcurrency.Value, minimumIterationTime: null, CancellationToken).ConfigureAwait(false);
     }
 }

EnumerableAsyncProcessor/RunnableProcessors/RateLimitedParallelAsyncProcessor.cs

@@ -1,4 +1,3 @@
-using EnumerableAsyncProcessor.Extensions;
 using EnumerableAsyncProcessor.RunnableProcessors.Abstract;
 using EnumerableAsyncProcessor.Validation;
 
@@ -17,9 +16,6 @@ internal RateLimitedParallelAsyncProcessor(int count, Func<Task> taskSelector, i
 
     internal override Task Process()
     {
-        // Task.Run guards the shared worker slots against synchronous code in user delegates
-        return TaskWrappers.InParallelAsync(_levelsOfParallelism,
-            taskWrapper => Task.Run(() => taskWrapper.Process(CancellationToken), CancellationToken),
-            CancellationToken);
+        return WorkerPool.ProcessAsync(TaskWrappers, _levelsOfParallelism, minimumIterationTime: null, CancellationToken);
     }
 }

EnumerableAsyncProcessor/RunnableProcessors/RateLimitedParallelAsyncProcessor_1.cs

@@ -1,4 +1,3 @@
-using EnumerableAsyncProcessor.Extensions;
 using EnumerableAsyncProcessor.RunnableProcessors.Abstract;
 using EnumerableAsyncProcessor.Validation;
 
@@ -17,9 +16,6 @@ internal RateLimitedParallelAsyncProcessor(IEnumerable<TInput> items, Func<TInpu
 
     internal override Task Process()
     {
-        // Task.Run guards the shared worker slots against synchronous code in user delegates
-        return TaskWrappers.InParallelAsync(_levelsOfParallelism,
-            taskWrapper => Task.Run(() => taskWrapper.Process(CancellationToken), CancellationToken),
-            CancellationToken);
+        return WorkerPool.ProcessAsync(TaskWrappers, _levelsOfParallelism, minimumIterationTime: null, CancellationToken);
     }
 }

EnumerableAsyncProcessor/RunnableProcessors/ResultProcessors/Abstract/ResultAbstractAsyncProcessor_1.cs

@@ -4,7 +4,7 @@ namespace EnumerableAsyncProcessor.RunnableProcessors.ResultProcessors.Abstract;
 
 public abstract class ResultAbstractAsyncProcessor<TOutput> : ResultAbstractAsyncProcessorBase<TOutput>
 {
-    protected readonly IReadOnlyList<ActionTaskWrapper<TOutput>> TaskWrappers;
+    protected readonly ActionTaskWrapper<TOutput>[] TaskWrappers;
 
     private readonly TaskCompletionSource<TOutput>[] _taskCompletionSources;
 

EnumerableAsyncProcessor/RunnableProcessors/ResultProcessors/Abstract/ResultAbstractAsyncProcessor_2.cs

@@ -4,7 +4,7 @@ namespace EnumerableAsyncProcessor.RunnableProcessors.ResultProcessors.Abstract;
 
 public abstract class ResultAbstractAsyncProcessor<TInput, TOutput> : ResultAbstractAsyncProcessorBase<TOutput>
 {
-    protected readonly IReadOnlyList<ItemTaskWrapper<TInput, TOutput>> TaskWrappers;
+    protected readonly ItemTaskWrapper<TInput, TOutput>[] TaskWrappers;
 
     private readonly TaskCompletionSource<TOutput>[] _taskCompletionSources;
 

EnumerableAsyncProcessor/RunnableProcessors/ResultProcessors/ResultParallelAsyncProcessor_1.cs

@@ -1,14 +1,20 @@
 using EnumerableAsyncProcessor.RunnableProcessors.ResultProcessors.Abstract;
+using EnumerableAsyncProcessor.Validation;
 
 namespace EnumerableAsyncProcessor.RunnableProcessors.ResultProcessors;
 
 public class ResultParallelAsyncProcessor<TOutput> : ResultAbstractAsyncProcessor<TOutput>
 {
     private readonly int? _maxConcurrency;
     private readonly bool _scheduleOnThreadPool;
-    
+
     internal ResultParallelAsyncProcessor(int count, Func<Task<TOutput>> taskSelector, CancellationTokenSource cancellationTokenSource, int? maxConcurrency = null, bool scheduleOnThreadPool = false) : base(count, taskSelector, cancellationTokenSource)
     {
+        if (maxConcurrency is { } concurrencyLimit)
+        {
+            ValidationHelper.ThrowIfNegativeOrZero(concurrencyLimit, nameof(maxConcurrency));
+        }
+
         _maxConcurrency = maxConcurrency;
         _scheduleOnThreadPool = scheduleOnThreadPool;
     }
@@ -38,38 +44,8 @@ await Task.WhenAll(TaskWrappers.Select(taskWrapper =>
             return;
         }
 
-        // Use semaphore for concurrency throttling
-        using var semaphore = new SemaphoreSlim(_maxConcurrency.Value, _maxConcurrency.Value);
-        
-        // Materialize tasks immediately to ensure they all start in parallel (up to concurrency limit)
-        var tasks = _scheduleOnThreadPool
-            ? // Use Task.Run to prevent synchronous code from blocking thread pool threads
-              TaskWrappers.Select(taskWrapper => Task.Run(async () =>
-        {
-            await semaphore.WaitAsync(CancellationToken).ConfigureAwait(false);
-            try
-            {
-                await taskWrapper.Process(CancellationToken).ConfigureAwait(false);
-            }
-            finally
-            {
-                semaphore.Release();
-            }
-              }, CancellationToken)).ToList()
-            : // Direct execution for maximum performance
-              TaskWrappers.Select(async taskWrapper =>
-              {
-                  await semaphore.WaitAsync(CancellationToken).ConfigureAwait(false);
-                  try
-                  {
-                      await taskWrapper.Process(CancellationToken).ConfigureAwait(false);
-                  }
-                  finally
-                  {
-                      semaphore.Release();
-                  }
-              }).ToList(); // Force immediate task creation
-
-        await Task.WhenAll(tasks).ConfigureAwait(false);
+        // Throttled processing runs on a fixed worker pool: P worker tasks instead of
+        // one queued task, closure and semaphore wait per item
+        await WorkerPool.ProcessAsync(TaskWrappers, _maxConcurrency.Value, minimumIterationTime: null, CancellationToken).ConfigureAwait(false);
     }
 }