perf: remove EnumerableAsyncProcessor dependency (#6465)

* perf: remove EnumerableAsyncProcessor dependency

Replace the EnumerableAsyncProcessor package with an internal
ParallelMap helper: a bounded-parallelism async map where a fixed
set of workers pulls items via an interlocked cursor. Results
preserve source order with no partitioner, channel, or per-item
task allocations, and sets below 8 items run sequentially to skip
task scheduling overhead entirely.

- Consolidate the duplicated sequential/parallel/streaming build
  paths in TestBuilderPipeline onto one shared per-metadata method
- Delete the fake-streaming IAsyncEnumerable path (it fully
  buffered before yielding) and its dead generic-resolution error
  helper
- Flatten build results eagerly with exact capacity instead of a
  lazy SelectMany that downstream code enumerated twice
- Rewrite the NuGet upload module loop without the library

* Address review: converge on ParallelMap, add tests, fix naming

- Route InvokePostResolutionEventsInParallelAsync through a new
  ParallelMap.ForEachParallelAsync (ValueTask action, no results
  array), deleting the #if NET8_0_OR_GREATER Parallel.ForEachAsync
  split whose netstandard2.0 fallback was unbounded
- Add dedicated ParallelMapTests: order preservation, threshold
  boundaries, exactly-once invocation, DOP cap, exception
  propagation and fail-fast cursor parking, cancellation
- Rename fake-streaming BuildTestsStreamingAsync to BuildTestsAsync
  and delete the dead zero-caller BuildTestsAsync(sessionId) overload

Author: thomhurst

Directory.Packages.props

@@ -12,7 +12,6 @@
     <PackageVersion Include="BenchmarkDotNet.Annotations" Version="0.15.8" />
     <PackageVersion Include="coverlet.collector" Version="10.0.1" />
     <PackageVersion Include="CliWrap" Version="3.10.2" />
-    <PackageVersion Include="EnumerableAsyncProcessor" Version="3.8.4" />
     <PackageVersion Include="FakeItEasy" Version="9.0.1" />
     <PackageVersion Include="FluentValidation.DependencyInjectionExtensions" Version="12.1.1" />
     <PackageVersion Include="FsCheck" Version="3.3.3" />

src/TUnit.Engine/Building/TestBuilderPipeline.cs

@@ -13,7 +13,6 @@
   </ItemGroup>
 
   <ItemGroup>
-    <PackageReference Include="EnumerableAsyncProcessor" />
     <PackageReference Include="Microsoft.Testing.Extensions.TrxReport.Abstractions" />
     <PackageReference Include="Microsoft.Testing.Platform" />
   </ItemGroup>

src/TUnit.Engine/TUnit.Engine.csproj

@@ -233,7 +233,7 @@ private async IAsyncEnumerable<AbstractExecutableTest> DiscoverTestsStreamAsync(
         using var cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
         cts.CancelAfter(EngineDefaults.DiscoveryTimeout);
 
-        var tests = await _testBuilderPipeline.BuildTestsStreamingAsync(testSessionId, buildingContext, metadataFilter: null, cts.Token).ConfigureAwait(false);
+        var tests = await _testBuilderPipeline.BuildTestsAsync(testSessionId, buildingContext, metadataFilter: null, cts.Token).ConfigureAwait(false);
 
         foreach (var test in tests)
         {
@@ -382,34 +382,12 @@ public async IAsyncEnumerable<AbstractExecutableTest> DiscoverTestsFullyStreamin
 
 
 
-    private async Task InvokePostResolutionEventsInParallelAsync(List<AbstractExecutableTest> allTests)
+    private Task InvokePostResolutionEventsInParallelAsync(List<AbstractExecutableTest> allTests)
     {
-        if (allTests.Count < Building.ParallelThresholds.MinItemsForParallel)
-        {
-            foreach (var test in allTests)
-            {
-                await _testBuilderPipeline.InvokePostResolutionEventsAsync(test).ConfigureAwait(false);
-            }
-            return;
-        }
-
-#if NET8_0_OR_GREATER
-        await Parallel.ForEachAsync(
+        return Utilities.ParallelMap.ForEachParallelAsync(
             allTests,
-            new ParallelOptions { MaxDegreeOfParallelism = Environment.ProcessorCount },
-            async (test, _) =>
-            {
-                await _testBuilderPipeline.InvokePostResolutionEventsAsync(test).ConfigureAwait(false);
-            }
-        ).ConfigureAwait(false);
-#else
-        var tasks = new Task[allTests.Count];
-        for (var i = 0; i < allTests.Count; i++)
-        {
-            tasks[i] = _testBuilderPipeline.InvokePostResolutionEventsAsync(allTests[i]).AsTask();
-        }
-        await Task.WhenAll(tasks).ConfigureAwait(false);
-#endif
+            test => _testBuilderPipeline.InvokePostResolutionEventsAsync(test),
+            Environment.ProcessorCount);
     }
 
     public IEnumerable<TestContext> GetCachedTestContexts()

src/TUnit.Engine/TestDiscoveryService.cs

@@ -0,0 +1,164 @@
+namespace TUnit.Engine.Utilities;
+
+/// <summary>
+/// Allocation-light bounded-parallelism async map.
+/// Results preserve source order. A fixed set of workers pulls items via an interlocked
+/// cursor, so no partitioner, channel, or per-item task is allocated.
+/// Sets smaller than <see cref="SequentialThreshold"/> run sequentially, where task
+/// scheduling overhead would exceed the parallelization benefit.
+/// </summary>
+internal static class ParallelMap
+{
+    /// <summary>
+    /// Minimum number of items before parallel processing is used.
+    /// </summary>
+    public const int SequentialThreshold = 8;
+
+    /// <summary>
+    /// Void counterpart of <see cref="SelectParallelAsync{TSource,TResult}"/> — same worker
+    /// model without the results array. Takes a ValueTask action so synchronously-completing
+    /// callees don't allocate a Task per item.
+    /// </summary>
+    public static async Task ForEachParallelAsync<TSource>(
+        IReadOnlyList<TSource> source,
+        Func<TSource, ValueTask> action,
+        int maxDegreeOfParallelism,
+        CancellationToken cancellationToken = default)
+    {
+        var count = source.Count;
+
+        if (count == 0)
+        {
+            return;
+        }
+
+        var workerCount = Math.Min(maxDegreeOfParallelism, count);
+
+        if (workerCount <= 1 || count < SequentialThreshold)
+        {
+            for (var i = 0; i < count; i++)
+            {
+                cancellationToken.ThrowIfCancellationRequested();
+                await action(source[i]).ConfigureAwait(false);
+            }
+
+            return;
+        }
+
+        var cursor = -1;
+
+        Func<Task> worker = WorkerAsync;
+        var workers = new Task[workerCount];
+        for (var w = 0; w < workerCount; w++)
+        {
+            workers[w] = Task.Run(worker, CancellationToken.None);
+        }
+
+        await Task.WhenAll(workers).ConfigureAwait(false);
+
+        return;
+
+        async Task WorkerAsync()
+        {
+            while (true)
+            {
+                var index = Interlocked.Increment(ref cursor);
+
+                if (index >= count)
+                {
+                    return;
+                }
+
+                cancellationToken.ThrowIfCancellationRequested();
+
+                try
+                {
+                    await action(source[index]).ConfigureAwait(false);
+                }
+                catch
+                {
+                    // Park the cursor so sibling workers stop pulling new items;
+                    // Task.WhenAll surfaces this fault once in-flight items finish.
+                    Volatile.Write(ref cursor, count);
+                    throw;
+                }
+            }
+        }
+    }
+
+    public static Task<TResult[]> SelectParallelAsync<TSource, TResult>(
+        IReadOnlyList<TSource> source,
+        Func<TSource, Task<TResult>> selector,
+        int maxDegreeOfParallelism,
+        CancellationToken cancellationToken = default)
+        => ForParallelAsync(source.Count, index => selector(source[index]), maxDegreeOfParallelism, cancellationToken);
+
+    public static async Task<TResult[]> ForParallelAsync<TResult>(
+        int count,
+        Func<int, Task<TResult>> selector,
+        int maxDegreeOfParallelism,
+        CancellationToken cancellationToken = default)
+    {
+        if (count == 0)
+        {
+            return [];
+        }
+
+        var results = new TResult[count];
+        var workerCount = Math.Min(maxDegreeOfParallelism, count);
+
+        if (workerCount <= 1 || count < SequentialThreshold)
+        {
+            for (var i = 0; i < count; i++)
+            {
+                cancellationToken.ThrowIfCancellationRequested();
+                results[i] = await selector(i).ConfigureAwait(false);
+            }
+
+            return results;
+        }
+
+        var cursor = -1;
+
+        // Task.Run rather than invoking WorkerAsync directly: a selector that completes
+        // synchronously would otherwise run the entire loop inline on this thread and
+        // serialize all the other workers.
+        Func<Task> worker = WorkerAsync;
+        var workers = new Task[workerCount];
+        for (var w = 0; w < workerCount; w++)
+        {
+            workers[w] = Task.Run(worker, CancellationToken.None);
+        }
+
+        await Task.WhenAll(workers).ConfigureAwait(false);
+
+        return results;
+
+        async Task WorkerAsync()
+        {
+            while (true)
+            {
+                var index = Interlocked.Increment(ref cursor);
+
+                if (index >= count)
+                {
+                    return;
+                }
+
+                cancellationToken.ThrowIfCancellationRequested();
+
+                try
+                {
+                    results[index] = await selector(index).ConfigureAwait(false);
+                }
+                catch
+                {
+                    // Park the cursor so sibling workers stop pulling new items;
+                    // Task.WhenAll surfaces this fault once in-flight items finish.
+                    Volatile.Write(ref cursor, count);
+                    throw;
+                }
+            }
+        }
+    }
+}