Skip to content

Pool bounded async-enumerable workers#358

Merged
thomhurst merged 2 commits into
mainfrom
agent/async-enumerable-worker-pool
Jul 21, 2026
Merged

Pool bounded async-enumerable workers#358
thomhurst merged 2 commits into
mainfrom
agent/async-enumerable-worker-pool

Conversation

@thomhurst

Copy link
Copy Markdown
Owner

Summary

  • replace bounded async-stream semaphore fan-out with a bounded Channel<T> and P persistent workers
  • cap ordered result buffering at P with Queue<Task<T>>, preserving input-order yields without List.RemoveAt(0) shifts
  • collect no-selector async streams directly instead of scheduling no-op tasks
  • remove redundant Task.Run(async () => await ...) wrappers from unbounded paths
  • add explicit ordering and source-backpressure regression tests

Impact

Bounded async-stream coordination now scales with configured concurrency instead of item count. Producers cannot run arbitrarily ahead, synchronous delegate prefixes run on worker threads, and result processors retain ordered streaming.

Validation

  • bounded processor tests: 17 passed
  • async-enumerable extension tests: 11 passed
  • searches confirm no bounded semaphore fan-out, RemoveAt(0), or redundant async Task.Run wrapper remains
  • dotnet test: 1,617 tests passed across net8.0, net9.0, and net10.0

Benchmark dependency

PR #353 owns the new benchmark project and is still open. The async-stream comparison should be added there after that project lands; this PR avoids duplicating the entire benchmark scaffold and creating a guaranteed conflict.

Closes #337

@greptile-apps

greptile-apps Bot commented Jul 21, 2026

Copy link
Copy Markdown

Greptile Summary

This PR replaces bounded async-enumerable fan-out with a persistent worker pool. The main changes are:

  • Uses bounded channels to limit queued work and source read-ahead.
  • Preserves input order with a bounded queue of result tasks.
  • Simplifies selector-free collection and unbounded task scheduling.
  • Adds ordering, backpressure, and concurrent-failure tests.

Confidence Score: 5/5

This looks safe to merge.

  • Concurrent worker failures now surface an original selector exception directly.
  • The new test covers overlapping failures in the bounded action path.
  • No blocking issue related to the earlier exception-handling failure remains.

Important Files Changed

Filename Overview
EnumerableAsyncProcessor/AsyncEnumerableWorkerPool.cs Adds bounded channel workers, ordered result buffering, cancellation cleanup, and direct propagation of the first captured action failure.
EnumerableAsyncProcessor.UnitTests/AsyncEnumerableProcessorTests.cs Adds tests for ordered results, source backpressure, and overlapping worker failures.
EnumerableAsyncProcessor/Extensions/AsyncEnumerableExtensions.cs Routes bounded result processing through the worker pool and simplifies collection and task scheduling.
EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/AsyncEnumerableParallelProcessor.cs Replaces bounded semaphore fan-out with the shared worker pool.
EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/ResultProcessors/ResultAsyncEnumerableParallelProcessor.cs Uses bounded worker-pool streaming while retaining the existing unbounded path.

Reviews (6): Last reviewed commit: "fix: preserve async stream exception typ..." | Re-trigger Greptile

Comment thread EnumerableAsyncProcessor/AsyncEnumerableWorkerPool.cs Outdated
@thomhurst
thomhurst force-pushed the agent/async-enumerable-worker-pool branch from 96512f5 to 801de90 Compare July 21, 2026 19:42
@thomhurst
thomhurst force-pushed the agent/async-enumerable-worker-pool branch from f76e249 to 07ce01a Compare July 21, 2026 19:51
@thomhurst
thomhurst force-pushed the agent/async-enumerable-worker-pool branch from 07ce01a to e97eab5 Compare July 21, 2026 19:59
Bounded async streams now use P persistent channel workers and capped ordered-result queues, keeping source read-ahead and coordination proportional to concurrency.

Refs #337
@thomhurst
thomhurst force-pushed the agent/async-enumerable-worker-pool branch from e97eab5 to c02c5a9 Compare July 21, 2026 20:03
@thomhurst
thomhurst merged commit b005d8b into main Jul 21, 2026
5 checks passed
@thomhurst
thomhurst deleted the agent/async-enumerable-worker-pool branch July 21, 2026 20:06
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

Perf: bring IAsyncEnumerable processors onto the worker-pool model

1 participant