Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion EnumerableAsyncProcessor/EnumerableAsyncProcessor.csproj
Original file line number Diff line number Diff line change
Expand Up @@ -17,8 +17,9 @@
<PropertyGroup>
<PackageLicenseExpression>MIT</PackageLicenseExpression>
<PackageReadmeFile>README.md</PackageReadmeFile>
<PackageIcon>icon.png</PackageIcon>
<Authors>Tom Longhurst</Authors>
<Description>Various Enumerable Async Processors - Batch / Parallel / Rate Limited / One at a time</Description>
<Description>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.</Description>
<RepositoryType>git</RepositoryType>
<PackageProjectUrl>https://github.com/thomhurst/EnumerableAsyncProcessor</PackageProjectUrl>
<RepositoryUrl>https://github.com/thomhurst/EnumerableAsyncProcessor</RepositoryUrl>
Expand All @@ -38,6 +39,7 @@
</AssemblyAttribute>

<None Include="$(MSBuildThisFileDirectory)..\README.md" Pack="true" PackagePath="\" />
<None Include="$(MSBuildThisFileDirectory)..\icon.png" Pack="true" PackagePath="\" />

<PackageReference Include="Microsoft.SourceLink.GitHub" Version="10.0.301" PrivateAssets="All"/>
<PackageReference Include="System.Threading.RateLimiting" Version="10.0.10" />
Expand Down
74 changes: 39 additions & 35 deletions EnumerableAsyncProcessor/Extensions/ParallelExtensions.cs
Original file line number Diff line number Diff line change
Expand Up @@ -2,25 +2,23 @@

namespace EnumerableAsyncProcessor.Extensions;

/// <summary>
/// Fire-and-await parallel processing over an <see cref="IEnumerable{T}"/> without building a processor object.
/// To collect results, use <c>SelectAsync(...).ProcessInParallel(...)</c> instead.
/// </summary>
public static class ParallelExtensions
{
public static Task InParallelAsync<TSource, TResult>(
this IEnumerable<TSource> source,
int levelOfParallelism,
Func<TSource, Task<TResult>> taskSelector)
{
return InParallelAsync(source, levelOfParallelism, taskSelector, CancellationToken.None);
}

public static Task InParallelAsync<TSource, TResult>(
this IEnumerable<TSource> source,
int levelOfParallelism,
Func<TSource, Task<TResult>> taskSelector,
CancellationToken cancellationToken)
{
return StartWorkers(source, levelOfParallelism, taskSelector, cancellationToken);
}

/// <summary>
/// Runs <paramref name="taskSelector"/> for every item on a fixed pool of workers.
/// </summary>
/// <typeparam name="TSource">The input item type.</typeparam>
/// <param name="source">The items to process. The sequence is materialized once before workers start.</param>
/// <param name="levelOfParallelism">Maximum concurrent operations. Zero or negative uses <see cref="Environment.ProcessorCount"/>.</param>
/// <param name="taskSelector">The async operation to perform on each item. Any result carried by a returned <c>Task&lt;T&gt;</c> is not collected.</param>
/// <returns>
/// A task that completes when every item has been processed. When multiple items fail, awaiting it
/// throws the first failure while <c>Task.Exception.InnerExceptions</c> carries every failure.
/// </returns>
public static Task InParallelAsync<TSource>(
this IEnumerable<TSource> source,
int levelOfParallelism,
Expand All @@ -29,6 +27,18 @@ public static Task InParallelAsync<TSource>(
return InParallelAsync(source, levelOfParallelism, taskSelector, CancellationToken.None);
}

/// <summary>
/// Runs <paramref name="taskSelector"/> for every item on a fixed pool of workers.
/// </summary>
/// <typeparam name="TSource">The input item type.</typeparam>
/// <param name="source">The items to process. The sequence is materialized once before workers start.</param>
/// <param name="levelOfParallelism">Maximum concurrent operations. Zero or negative uses <see cref="Environment.ProcessorCount"/>.</param>
/// <param name="taskSelector">The async operation to perform on each item. Any result carried by a returned <c>Task&lt;T&gt;</c> is not collected.</param>
/// <param name="cancellationToken">Stops claiming new items when cancelled. The returned task then completes as canceled, unless items had already failed - their exceptions take precedence.</param>
/// <returns>
/// A task that completes when every item has been processed. When multiple items fail, awaiting it
/// throws the first failure while <c>Task.Exception.InnerExceptions</c> carries every failure.
/// </returns>
public static Task InParallelAsync<TSource>(
this IEnumerable<TSource> source,
int levelOfParallelism,
Expand All @@ -38,24 +48,18 @@ public static Task InParallelAsync<TSource>(
return StartWorkers(source, levelOfParallelism, taskSelector, cancellationToken);
}

// Overloads for CPU-bound processing
public static Task InParallelAsync<TSource, TResult>(
this IEnumerable<TSource> source,
int levelOfParallelism,
Func<TSource, TResult> taskSelector,
CancellationToken cancellationToken = default)
{
return StartWorkers(
source,
levelOfParallelism,
item =>
{
_ = taskSelector(item);
return Task.CompletedTask;
},
cancellationToken);
}

/// <summary>
/// Runs a synchronous, CPU-bound <paramref name="taskSelector"/> for every item on a fixed pool of thread-pool workers.
/// </summary>
/// <typeparam name="TSource">The input item type.</typeparam>
/// <param name="source">The items to process. The sequence is materialized once before workers start.</param>
/// <param name="levelOfParallelism">Maximum concurrent operations. Zero or negative uses <see cref="Environment.ProcessorCount"/>.</param>
/// <param name="taskSelector">The synchronous operation to perform on each item.</param>
/// <param name="cancellationToken">Stops claiming new items when cancelled. The returned task then completes as canceled, unless items had already failed - their exceptions take precedence.</param>
/// <returns>
/// A task that completes when every item has been processed. When multiple items fail, awaiting it
/// throws the first failure while <c>Task.Exception.InnerExceptions</c> carries every failure.
/// </returns>
public static Task InParallelAsync<TSource>(
this IEnumerable<TSource> source,
int levelOfParallelism,
Expand Down
53 changes: 37 additions & 16 deletions EnumerableAsyncProcessor/Interfaces/IAsyncProcessor.cs
Original file line number Diff line number Diff line change
@@ -1,24 +1,45 @@
using System.Runtime.CompilerServices;
using System.Runtime.CompilerServices;

namespace EnumerableAsyncProcessor.Interfaces;

/// <summary>
/// Represents an async processor that performs operations without returning results.
/// </summary>
/// <remarks>
/// 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.
/// </remarks>
public interface IAsyncProcessor : IAsyncDisposable, IDisposable
{
/**
* <summary>
* A collection of all the asynchronous Tasks, which could be pending or complete.
* </summary>
*/
IEnumerable<Task> GetEnumerableTasks();
/// <summary>
/// Gets a collection of all the asynchronous Tasks, which could be pending or complete.
/// Tasks appear in source order.
/// </summary>
/// <returns>An enumerable of tasks that may be running, completed, or faulted.</returns>
IEnumerable<Task> GetEnumerableTasks();

TaskAwaiter GetAwaiter();
/// <summary>
/// Gets a task awaiter for awaiting completion of all operations directly.
/// </summary>
/// <returns>A task awaiter that completes when all processing is done.</returns>
TaskAwaiter GetAwaiter();

Task WaitAsync();
/// <summary>
/// Gets a task that completes when all operations have finished.
/// </summary>
/// <returns>
/// A task that completes when all processing is done. When multiple items fail, awaiting it
/// throws the first failure while <c>Task.Exception.InnerExceptions</c> carries every failure.
/// </returns>
Task WaitAsync();

/**
* <summary>
* Try to cancel all un-finished tasks
* </summary>
*/
void CancelAll();
}
/// <summary>
/// Attempts to cancel all unfinished tasks.
/// </summary>
/// <remarks>
/// This method is called automatically during disposal but can be called manually
/// if you need to cancel processing before disposal.
/// </remarks>
void CancelAll();
}
3 changes: 0 additions & 3 deletions EnumerableAsyncProcessor/PublicAPI.Shipped.txt
Original file line number Diff line number Diff line change
Expand Up @@ -169,9 +169,6 @@ static EnumerableAsyncProcessor.Extensions.EnumerableExtensions.SelectManyAsync<
static EnumerableAsyncProcessor.Extensions.EnumerableExtensions.SelectManyAsync<T, TOutput>(this System.Collections.Generic.IEnumerable<T>! items, System.Func<T, System.Threading.Tasks.Task<System.Collections.Generic.IEnumerable<TOutput>!>!>! selector, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Collections.Generic.IAsyncEnumerable<TOutput>!
static EnumerableAsyncProcessor.Extensions.EnumerableExtensions.SelectManyAsync<T, TOutput>(this System.Collections.Generic.IEnumerable<T>! items, System.Func<T, System.Threading.Tasks.Task<TOutput[]!>!>! selector, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Collections.Generic.IAsyncEnumerable<TOutput>!
static EnumerableAsyncProcessor.Extensions.EnumerableExtensions.ToAsyncProcessorBuilder<T>(this System.Collections.Generic.IEnumerable<T>! items) -> EnumerableAsyncProcessor.Builders.ItemAsyncProcessorBuilder<T>!
static EnumerableAsyncProcessor.Extensions.ParallelExtensions.InParallelAsync<TSource, TResult>(this System.Collections.Generic.IEnumerable<TSource>! source, int levelOfParallelism, System.Func<TSource, System.Threading.Tasks.Task<TResult>!>! taskSelector) -> System.Threading.Tasks.Task!
static EnumerableAsyncProcessor.Extensions.ParallelExtensions.InParallelAsync<TSource, TResult>(this System.Collections.Generic.IEnumerable<TSource>! source, int levelOfParallelism, System.Func<TSource, System.Threading.Tasks.Task<TResult>!>! taskSelector, System.Threading.CancellationToken cancellationToken) -> System.Threading.Tasks.Task!
static EnumerableAsyncProcessor.Extensions.ParallelExtensions.InParallelAsync<TSource, TResult>(this System.Collections.Generic.IEnumerable<TSource>! source, int levelOfParallelism, System.Func<TSource, TResult>! taskSelector, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task!
static EnumerableAsyncProcessor.Extensions.ParallelExtensions.InParallelAsync<TSource>(this System.Collections.Generic.IEnumerable<TSource>! source, int levelOfParallelism, System.Action<TSource>! taskSelector, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task!
static EnumerableAsyncProcessor.Extensions.ParallelExtensions.InParallelAsync<TSource>(this System.Collections.Generic.IEnumerable<TSource>! source, int levelOfParallelism, System.Func<TSource, System.Threading.Tasks.Task!>! taskSelector) -> System.Threading.Tasks.Task!
static EnumerableAsyncProcessor.Extensions.ParallelExtensions.InParallelAsync<TSource>(this System.Collections.Generic.IEnumerable<TSource>! source, int levelOfParallelism, System.Func<TSource, System.Threading.Tasks.Task!>! taskSelector, System.Threading.CancellationToken cancellationToken) -> System.Threading.Tasks.Task!
3 changes: 2 additions & 1 deletion README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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<TSource, TResult>` and `Action<TSource>` 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<TSource>` 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<TSource, Task<TResult>>` and `Func<TSource, TResult>` 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<T>` (they bind to the `Func<TSource, Task>` 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<T>` 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<T>` 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.
Expand Down
Binary file added icon.png
Loading
Sorry, something went wrong. Reload?
Sorry, we cannot display this file.
Sorry, this file is invalid so it cannot be displayed.