Use worker pool for InParallelAsync

[thomhurst]

Summary

• replace eager per-item semaphore fan-out with a fixed worker pool
• materialize the source once and keep coordination work proportional to configured parallelism
• run synchronous delegates on persistent Task.Run workers so CPU-bound overloads execute concurrently
• drain all items after failures and preserve every fault on the returned task

Impact

Large inputs no longer allocate one state machine, task, and semaphore waiter per item. Both asynchronous and synchronous overloads now share the same bounded worker model.

Validation

• dotnet test
• 1,620 tests passed across net8.0, net9.0, and net10.0
• added coverage for synchronous parallelism, single source enumeration, and multi-failure draining

Closes #338

[greptile-apps]

Greptile Summary

This PR replaces per-item semaphore fan-out with a bounded worker pool. The main changes are:

• Materializes the source once before starting workers.
• Runs asynchronous and synchronous delegates through the same fixed worker model.
• Drains remaining items after failures and collects all task faults.
• Adds tests for CPU parallelism, fault draining, and single enumeration.

Confidence Score: 5/5

This looks safe to merge.

• No blocking issues found in the changed code.
• Worker completion waits for every worker and preserves collected failures.
• The new APIs are available across the repository's target frameworks.

Important Files Changed

Filename	Overview
EnumerableAsyncProcessor/Extensions/ParallelExtensions.cs	Replaces semaphore-based fan-out with bounded workers, atomic item claiming, cancellation tracking, and collected fault completion.
EnumerableAsyncProcessor.UnitTests/ParallelExtensionsTests.cs	Adds tests for synchronous parallel execution, processing after faults, fault preservation, and one-time source enumeration.

Flowchart

%%{init: {'theme': 'neutral'}}%%
flowchart TD
    A[InParallelAsync] --> B[Materialize source]
    B --> C{Any items?}
    C -- No --> D[Complete successfully]
    C -- Yes --> E[Start bounded workers]
    E --> F[Claim next index atomically]
    F --> G{Work available?}
    G -- Yes --> H[Run and await selector]
    H -->|Success| F
    H -->|Fault| I[Queue exceptions]
    I --> F
    H -->|Canceled| J[Record cancellation]
    J --> F
    G -- No --> K[Worker exits]
    K --> L{Last worker?}
    L -- No --> K
    L -- Yes --> M{Faults collected?}
    M -- Yes --> N[Complete faulted]
    M -- No --> O{Cancellation observed?}
    O -- Yes --> P[Complete canceled]
    O -- No --> D

Loading

%%{init: {'theme': 'base', 'themeVariables': {"darkMode": true, "background": "#0d1117", "primaryColor": "#21262d", "primaryTextColor": "#e6edf3", "primaryBorderColor": "#8b949e", "lineColor": "#8b949e", "textColor": "#e6edf3", "edgeLabelBackground": "#161b22", "actorBkg": "#21262d", "actorBorder": "#8b949e", "actorTextColor": "#e6edf3", "actorLineColor": "#8b949e", "signalColor": "#8b949e", "signalTextColor": "#e6edf3", "noteBkgColor": "#373320", "noteBorderColor": "#d4a72c", "noteTextColor": "#f0e6c0", "labelBoxBkgColor": "#21262d", "labelBoxBorderColor": "#8b949e", "labelTextColor": "#e6edf3", "loopTextColor": "#e6edf3", "activationBkgColor": "#30363d", "activationBorderColor": "#8b949e"}}}%%
flowchart TD
    A[InParallelAsync] --> B[Materialize source]
    B --> C{Any items?}
    C -- No --> D[Complete successfully]
    C -- Yes --> E[Start bounded workers]
    E --> F[Claim next index atomically]
    F --> G{Work available?}
    G -- Yes --> H[Run and await selector]
    H -->|Success| F
    H -->|Fault| I[Queue exceptions]
    I --> F
    H -->|Canceled| J[Record cancellation]
    J --> F
    G -- No --> K[Worker exits]
    K --> L{Last worker?}
    L -- No --> K
    L -- Yes --> M{Faults collected?}
    M -- Yes --> N[Complete faulted]
    M -- No --> O{Cancellation observed?}
    O -- Yes --> P[Complete canceled]
    O -- No --> D

Loading

_{Reviews (1): Last reviewed commit: "perf: use worker pool in InParallelAsync" | Re-trigger Greptile}