perf: remove EnumerableAsyncProcessor dependency#6465
Conversation
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
Greptile SummaryThis PR replaces
Confidence Score: 5/5This looks safe to merge.
Important Files Changed
Reviews (2): Last reviewed commit: "Address review: converge on ParallelMap,..." | Re-trigger Greptile |
Before/after performanceEnd-to-end discovery —
Source-gen std dev ~60 ms per variant — the gap is ~9 sigma, well outside noise. Wall-clock includes constant process startup/JIT, so the discovery-phase improvement is larger than the percentage suggests. Primitive micro-benchmark —
The end-to-end win comes from the primitive swap plus deleting the fake-streaming path (per-metadata async-iterator state machines and triple list copies). |
ReviewSolid, well-motivated cleanup — dropping a transitive dependency, and the One architectural concern worth resolving before merge, plus two smaller follow-ups.
|
- 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
|
Addressed all three review points in 5b4004e: 1. ParallelMap is now the single bounded-parallelism strategy. 2. Dedicated 3. Naming fixed. Verified: engine builds clean on all TFMs, |
|
Re-profiled at head 5b4004e (post-review changes:
(Noisier machine this session — two spikes hit both variants equally thanks to interleaving; medians robust.) Primitive micro-benchmark unchanged within noise: ~9–12x faster, ~4–9x fewer allocations vs |
![claude[bot]](https://avatars.githubusercontent.com/in/1236702?s=60&v=4){kind=small}
There was a problem hiding this comment.
Review
Re-reviewed the latest state (commit 5b4004e). This addresses all three points from the prior review round:
- ✅
ParallelMapis now the single bounded-parallelism strategy —TestDiscoveryService.InvokePostResolutionEventsInParallelAsyncroutes throughParallelMap.ForEachParallelAsync, and the old#if NET8_0_OR_GREATER/Parallel.ForEachAsyncvs. unboundedTask.WhenAllsplit is gone. - ✅
ParallelMapTests.csgives the new primitive direct coverage: order preservation,SequentialThresholdboundary, DOP-cap, fail-fast cursor parking, cancellation. - ✅
BuildTestsStreamingAsync→BuildTestsAsyncnaming fixed, and the dead zero-caller overload was deleted rather than kept around.
Nothing blocking. One design suggestion for ParallelMap.cs:
ForEachParallelAsync and ForParallelAsync duplicate the tricky part of the primitive
Both methods independently implement the same interlocked-cursor worker loop, including the "park the cursor on fault so siblings stop pulling" trick (Interlocked.Increment + Volatile.Write on the same field). That's the one piece of this type that's genuinely hard to get right, and it now exists as two separate copies that could silently drift apart under a future edit (e.g. someone fixes a cancellation edge case in one and forgets the other).
It can be factored into one private core loop parameterized by a per-index step:
private static async Task RunWorkersAsync(int count, int workerCount, CancellationToken ct, Func<int, ValueTask> step)
{
// the existing cursor/Task.Run/park-on-fault loop, unchanged
}
public static Task ForEachParallelAsync<T>(IReadOnlyList<T> source, Func<T, ValueTask> action, int dop, CancellationToken ct = default)
{
/* count==0 / sequential-threshold short-circuits stay here, same as today */
return RunWorkersAsync(source.Count, Math.Min(dop, source.Count), ct, i => action(source[i]));
}
public static async Task<TResult[]> ForParallelAsync<TResult>(int count, Func<int, Task<TResult>> selector, int dop, CancellationToken ct = default)
{
var results = new TResult[count];
await RunWorkersAsync(count, Math.Min(dop, count), ct, async i => results[i] = await selector(i)).ConfigureAwait(false);
return results;
}This keeps the allocation profile identical (ValueTask step, no per-item Task, no results array for the void path) while leaving exactly one implementation of the lock-free cursor logic to reason about and test. Not blocking — the current duplication is well-tested on the SelectParallelAsync/ForParallelAsync side, and ForEachParallelAsync's loop is a straight copy of it — but worth doing before this primitive picks up a third callsite or a third variant.
Otherwise: the TestBuilderPipeline consolidation is a real simplification (the eager-capacity flatten, the single BuildTestsForMetadataAsync per-item method replacing three divergent paths), and the plain sequential for loop in UploadToNuGetModule is the right call over forcing a DOP=1 ParallelMap call for an inherently one-at-a-time operation. Good cleanup overall.
1 participant
Summary
Removes the
EnumerableAsyncProcessorpackage dependency fromTUnit.Engine(one fewer transitive dependency for every TUnit user) and replaces it with an internal, allocation-lightParallelMaphelper purpose-built for the discovery hot path.The replacement (
ParallelMap)min(DOP, count)) pulls items via an interlocked cursor — no partitioner, no channels, no per-itemTask, one results array, one shared closure.Task.WhenAllsurfaces the fault.netstandard2.0(whereParallel.ForEachAsyncis unavailable).Cleanups enabled by the rewrite
TestBuilderPipeline's three build paths (sequential small-set branch, parallel branch, "streaming" branch) collapsed onto one shared per-metadata method — the streaming path was fake-streaming (fully buffered before yielding) and its dynamic-test branch was a ~120-line copy ofGenerateDynamicTests.CreateFailedTestForGenericResolutionError+ unreachable resolution-error scaffolding deleted.SelectManythatTestRegistryenumerated twice.UploadToNuGetModulepush loop rewritten as a plain sequentialfor(same one-at-a-time semantics).Net: -138 lines (163+ / 301-).
Behavior notes
Environment.ProcessorCount, consistent with the non-streaming path.Task.Runworker overhead.Test plan
TUnit.Engine+TUnit.Pipelinebuild clean on all TFMs, 0 warningsTUnit.UnitTests: 259/259 passTUnit.TestProjectdynamic-test filter: pass in both source-gen and reflection modesTUnit.TestProjectrepeat-attribute tests: passEnumerableAsyncProcessorreferences