diff --git a/EnumerableAsyncProcessor/EnumerableAsyncProcessor.csproj b/EnumerableAsyncProcessor/EnumerableAsyncProcessor.csproj index 3c58fa3..017c4c7 100644 --- a/EnumerableAsyncProcessor/EnumerableAsyncProcessor.csproj +++ b/EnumerableAsyncProcessor/EnumerableAsyncProcessor.csproj @@ -17,8 +17,9 @@ MIT README.md + icon.png Tom Longhurst - Various Enumerable Async Processors - Batch / Parallel / Rate Limited / One at a time + Process async tasks over IEnumerable and IAsyncEnumerable sources with controlled concurrency: one at a time, batched, bounded parallel, timed rate limited (e.g. requests per second), or fully parallel. Fluent API with streaming results, exception fidelity, cancellation and disposal support. git https://github.com/thomhurst/EnumerableAsyncProcessor https://github.com/thomhurst/EnumerableAsyncProcessor @@ -38,6 +39,7 @@ + diff --git a/EnumerableAsyncProcessor/Extensions/ParallelExtensions.cs b/EnumerableAsyncProcessor/Extensions/ParallelExtensions.cs index d203211..03ca479 100644 --- a/EnumerableAsyncProcessor/Extensions/ParallelExtensions.cs +++ b/EnumerableAsyncProcessor/Extensions/ParallelExtensions.cs @@ -2,25 +2,23 @@ namespace EnumerableAsyncProcessor.Extensions; +/// +/// Fire-and-await parallel processing over an without building a processor object. +/// To collect results, use SelectAsync(...).ProcessInParallel(...) instead. +/// public static class ParallelExtensions { - public static Task InParallelAsync( - this IEnumerable source, - int levelOfParallelism, - Func> taskSelector) - { - return InParallelAsync(source, levelOfParallelism, taskSelector, CancellationToken.None); - } - - public static Task InParallelAsync( - this IEnumerable source, - int levelOfParallelism, - Func> taskSelector, - CancellationToken cancellationToken) - { - return StartWorkers(source, levelOfParallelism, taskSelector, cancellationToken); - } - + /// + /// Runs for every item on a fixed pool of workers. + /// + /// The input item type. + /// The items to process. The sequence is materialized once before workers start. + /// Maximum concurrent operations. Zero or negative uses . + /// The async operation to perform on each item. Any result carried by a returned Task<T> is not collected. + /// + /// A task that completes when every item has been processed. When multiple items fail, awaiting it + /// throws the first failure while Task.Exception.InnerExceptions carries every failure. + /// public static Task InParallelAsync( this IEnumerable source, int levelOfParallelism, @@ -29,6 +27,18 @@ public static Task InParallelAsync( return InParallelAsync(source, levelOfParallelism, taskSelector, CancellationToken.None); } + /// + /// Runs for every item on a fixed pool of workers. + /// + /// The input item type. + /// The items to process. The sequence is materialized once before workers start. + /// Maximum concurrent operations. Zero or negative uses . + /// The async operation to perform on each item. Any result carried by a returned Task<T> is not collected. + /// Stops claiming new items when cancelled. The returned task then completes as canceled, unless items had already failed - their exceptions take precedence. + /// + /// A task that completes when every item has been processed. When multiple items fail, awaiting it + /// throws the first failure while Task.Exception.InnerExceptions carries every failure. + /// public static Task InParallelAsync( this IEnumerable source, int levelOfParallelism, @@ -38,24 +48,18 @@ public static Task InParallelAsync( return StartWorkers(source, levelOfParallelism, taskSelector, cancellationToken); } - // Overloads for CPU-bound processing - public static Task InParallelAsync( - this IEnumerable source, - int levelOfParallelism, - Func taskSelector, - CancellationToken cancellationToken = default) - { - return StartWorkers( - source, - levelOfParallelism, - item => - { - _ = taskSelector(item); - return Task.CompletedTask; - }, - cancellationToken); - } - + /// + /// Runs a synchronous, CPU-bound for every item on a fixed pool of thread-pool workers. + /// + /// The input item type. + /// The items to process. The sequence is materialized once before workers start. + /// Maximum concurrent operations. Zero or negative uses . + /// The synchronous operation to perform on each item. + /// Stops claiming new items when cancelled. The returned task then completes as canceled, unless items had already failed - their exceptions take precedence. + /// + /// A task that completes when every item has been processed. When multiple items fail, awaiting it + /// throws the first failure while Task.Exception.InnerExceptions carries every failure. + /// public static Task InParallelAsync( this IEnumerable source, int levelOfParallelism, diff --git a/EnumerableAsyncProcessor/Interfaces/IAsyncProcessor.cs b/EnumerableAsyncProcessor/Interfaces/IAsyncProcessor.cs index c9bc51c..099e42a 100644 --- a/EnumerableAsyncProcessor/Interfaces/IAsyncProcessor.cs +++ b/EnumerableAsyncProcessor/Interfaces/IAsyncProcessor.cs @@ -1,24 +1,45 @@ -using System.Runtime.CompilerServices; +using System.Runtime.CompilerServices; namespace EnumerableAsyncProcessor.Interfaces; +/// +/// Represents an async processor that performs operations without returning results. +/// +/// +/// This interface implements both IDisposable and IAsyncDisposable. Processors should be properly disposed +/// to ensure cleanup of internal resources and cancellation of running tasks. +/// Use 'await using var processor = ...' for automatic disposal, or call DisposeAsync() manually. +/// public interface IAsyncProcessor : IAsyncDisposable, IDisposable { - /** - * - * A collection of all the asynchronous Tasks, which could be pending or complete. - * - */ - IEnumerable GetEnumerableTasks(); + /// + /// Gets a collection of all the asynchronous Tasks, which could be pending or complete. + /// Tasks appear in source order. + /// + /// An enumerable of tasks that may be running, completed, or faulted. + IEnumerable GetEnumerableTasks(); - TaskAwaiter GetAwaiter(); + /// + /// Gets a task awaiter for awaiting completion of all operations directly. + /// + /// A task awaiter that completes when all processing is done. + TaskAwaiter GetAwaiter(); - Task WaitAsync(); + /// + /// Gets a task that completes when all operations have finished. + /// + /// + /// A task that completes when all processing is done. When multiple items fail, awaiting it + /// throws the first failure while Task.Exception.InnerExceptions carries every failure. + /// + Task WaitAsync(); - /** - * - * Try to cancel all un-finished tasks - * - */ - void CancelAll(); -} \ No newline at end of file + /// + /// Attempts to cancel all unfinished tasks. + /// + /// + /// This method is called automatically during disposal but can be called manually + /// if you need to cancel processing before disposal. + /// + void CancelAll(); +} diff --git a/EnumerableAsyncProcessor/PublicAPI.Shipped.txt b/EnumerableAsyncProcessor/PublicAPI.Shipped.txt index ab34a09..a2ceb8e 100644 --- a/EnumerableAsyncProcessor/PublicAPI.Shipped.txt +++ b/EnumerableAsyncProcessor/PublicAPI.Shipped.txt @@ -169,9 +169,6 @@ static EnumerableAsyncProcessor.Extensions.EnumerableExtensions.SelectManyAsync< static EnumerableAsyncProcessor.Extensions.EnumerableExtensions.SelectManyAsync(this System.Collections.Generic.IEnumerable! items, System.Func!>!>! selector, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Collections.Generic.IAsyncEnumerable! static EnumerableAsyncProcessor.Extensions.EnumerableExtensions.SelectManyAsync(this System.Collections.Generic.IEnumerable! items, System.Func!>! selector, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Collections.Generic.IAsyncEnumerable! static EnumerableAsyncProcessor.Extensions.EnumerableExtensions.ToAsyncProcessorBuilder(this System.Collections.Generic.IEnumerable! items) -> EnumerableAsyncProcessor.Builders.ItemAsyncProcessorBuilder! -static EnumerableAsyncProcessor.Extensions.ParallelExtensions.InParallelAsync(this System.Collections.Generic.IEnumerable! source, int levelOfParallelism, System.Func!>! taskSelector) -> System.Threading.Tasks.Task! -static EnumerableAsyncProcessor.Extensions.ParallelExtensions.InParallelAsync(this System.Collections.Generic.IEnumerable! source, int levelOfParallelism, System.Func!>! taskSelector, System.Threading.CancellationToken cancellationToken) -> System.Threading.Tasks.Task! -static EnumerableAsyncProcessor.Extensions.ParallelExtensions.InParallelAsync(this System.Collections.Generic.IEnumerable! source, int levelOfParallelism, System.Func! taskSelector, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task! static EnumerableAsyncProcessor.Extensions.ParallelExtensions.InParallelAsync(this System.Collections.Generic.IEnumerable! source, int levelOfParallelism, System.Action! taskSelector, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task! static EnumerableAsyncProcessor.Extensions.ParallelExtensions.InParallelAsync(this System.Collections.Generic.IEnumerable! source, int levelOfParallelism, System.Func! taskSelector) -> System.Threading.Tasks.Task! static EnumerableAsyncProcessor.Extensions.ParallelExtensions.InParallelAsync(this System.Collections.Generic.IEnumerable! source, int levelOfParallelism, System.Func! taskSelector, System.Threading.CancellationToken cancellationToken) -> System.Threading.Tasks.Task! diff --git a/README.md b/README.md index 6047321..09d75e9 100644 --- a/README.md +++ b/README.md @@ -35,7 +35,8 @@ Version 4 is a major release. Review these source and behaviour changes before u - **Arbitrary upper limits were removed.** Task counts and batch sizes may exceed 10,000, and time windows may exceed 24 hours. Validity checks remain: counts, batch sizes, concurrency, and permit counts must be positive; time windows cannot be negative. - **Validation is consistent and eager.** Invalid `maxConcurrency`, parallelism, batch, and timed-rate arguments now throw while the processor is built for both action and result variants. - **`TaskWrapper` types and processor plumbing are internal.** The `ActionTaskWrapper`/`ItemTaskWrapper` structs and the previously `protected` members of the abstract processor base classes (`TaskWrappers`, `EnumerableTaskCompletionSources`, `CancellationToken`, constructors) were implementation details that could not be meaningfully used outside the library, and are no longer public. -- **Synchronous `InParallelAsync` delegates now run concurrently.** The `Func` and `Action` overloads use thread-pool workers instead of executing delegates sequentially on the caller. Do not depend on their previous thread affinity or execution order. +- **Synchronous `InParallelAsync` delegates now run concurrently.** The `Action` overload uses thread-pool workers instead of executing delegates sequentially on the caller. Do not depend on its previous thread affinity or execution order. +- **The result-selector `InParallelAsync` overloads were removed.** The `Func>` and `Func` overloads returned a plain `Task` and silently discarded every result. Use `SelectAsync(...).ProcessInParallel(...)` to collect results. Fire-and-await callers can keep non-async lambdas returning `Task` (they bind to the `Func` overload), but async lambdas that return a value and explicitly typed result-returning delegates must be rewritten to discard the result (e.g. `async x => { _ = await GetAsync(x); }`). - **An already-cancelled token yields cancelled tasks, not `ArgumentException`.** Building a processor with a token that is already cancelled (or that cancels between building and the terminal call) now produces a processor whose per-item tasks are cancelled; `WaitAsync`/`GetResultsAsync` throw `OperationCanceledException`. Previously this threw `ArgumentException` from an internal constructor. - **`IAsyncEnumerable` processors preserve every failure.** The `Task` returned by `ExecuteAsync()` now carries all failures via `Task.Exception.InnerExceptions` (awaiting still throws the first), matching the `IEnumerable` processors. Previously only the first failure was observable and the rest were lost. - **`ExecuteAsync()` enforces single use.** Calling it a second time throws `InvalidOperationException`; calling it after disposal throws `ObjectDisposedException`. Previously a second call failed with a confusing `ObjectDisposedException` from internal plumbing. diff --git a/icon.png b/icon.png new file mode 100644 index 0000000..3191058 Binary files /dev/null and b/icon.png differ