Problem
PR #332 moved the array-based processors to a fixed worker pool, but the IAsyncEnumerable processors and extensions still use the older per-item machinery (findings from the perf audit):
AsyncEnumerableParallelProcessor / ResultAsyncEnumerableParallelProcessor (rate-limited paths): per item, one closure + one Task.Run task + a semaphore wait — ~300-400 B and two scheduler hops each.
ResultAsyncEnumerableParallelProcessor: head-drain via List.RemoveAt(0) — O(N²) reference shifts at scale; a Queue<Task<TOutput>> is O(1) and only the head is ever inspected.
AsyncEnumerableExtensions.ProcessInParallel no-selector overloads: pay Task.Run + semaphore per item to do nothing — it is a ToListAsync costing 10-100x what it should; a plain await foreach collect suffices.
Task.Run(async () => await f(x)) double state machine in 3 sites — Task.Run(() => f(x)) unwraps identically.
Proposal
- Streaming rate-limited paths: P persistent workers reading a bounded
Channel<TInput> (same shape as WorkerPool, adapted to streaming) — preserves the order-preserving yield semantics of the result processor.
Queue for the head-drain.
- Direct collect for no-selector overloads; drop redundant async lambdas.
Acceptance criteria
- Streaming behaviour (ordering, cancellation, backpressure) covered by tests before refactoring.
- Benchmark comparison in the benchmarks project (see benchmarks issue).
Problem
PR #332 moved the array-based processors to a fixed worker pool, but the
IAsyncEnumerableprocessors and extensions still use the older per-item machinery (findings from the perf audit):AsyncEnumerableParallelProcessor/ResultAsyncEnumerableParallelProcessor(rate-limited paths): per item, one closure + oneTask.Runtask + a semaphore wait — ~300-400 B and two scheduler hops each.ResultAsyncEnumerableParallelProcessor: head-drain viaList.RemoveAt(0)— O(N²) reference shifts at scale; aQueue<Task<TOutput>>is O(1) and only the head is ever inspected.AsyncEnumerableExtensions.ProcessInParallelno-selector overloads: payTask.Run+ semaphore per item to do nothing — it is aToListAsynccosting 10-100x what it should; a plainawait foreachcollect suffices.Task.Run(async () => await f(x))double state machine in 3 sites —Task.Run(() => f(x))unwraps identically.Proposal
Channel<TInput>(same shape asWorkerPool, adapted to streaming) — preserves the order-preserving yield semantics of the result processor.Queuefor the head-drain.Acceptance criteria