Skip to content

Commit 2356216

Browse files
committed
refactor!: freeze a clean v4 public API surface
Now-or-never cleanup before the 4.0.0 contract freezes: - Internalize the ActionTaskWrapper/ItemTaskWrapper structs and demote the abstract processor bases' protected plumbing (TaskWrappers, EnumerableTaskCompletionSources, CancellationToken, constructors, DisposeAsyncCore) to private protected - none of it was usable outside the assembly because Process() is internal abstract. - Seal every leaf processor and builder class. - Move IAsyncEnumerableProcessor to EnumerableAsyncProcessor.Interfaces and make both variants IAsyncDisposable/IDisposable. The six async-enumerable processors now dispose their linked CancellationTokenSource when ExecuteAsync completes (previously leaked a registration on the caller's token) and support explicit disposal. - Honor scheduleOnThreadPool on the bounded parallel paths (it was silently ignored whenever maxConcurrency was set). - Remove the parameterized no-selector IAsyncEnumerable ProcessInParallel overloads whose maxConcurrency/scheduleOnThreadPool did nothing. - Keep binary compatibility with assemblies compiled against v3, notably TUnit.Engine, which this repo's own test runner depends on: restore the parameterless ProcessInParallel(items, ct) collect overload and add ProcessInParallel(int) forwarders on the four enumerable builders. V3BinaryCompatibilityTests pins the exact signatures; this also fixes --treenode-filter discovery, broken since the v4 API consolidation. - Packaging: generate XML docs (IntelliSense was missing from the package), mark AOT-compatible, merge duplicate metadata groups. - Docs: correct stale CLAUDE.md/README claims (RateLimitedParallel strategy, minimumIterationTime model, disposal contract) and expand the migration guide.
1 parent 96e1c26 commit 2356216

49 files changed

Lines changed: 486 additions & 308 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

CLAUDE.md

Lines changed: 6 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,8 @@ dotnet run --project EnumerableAsyncProcessor.UnitTests -f net10.0 -- --treenode
2929

3030
TUnit test projects compile to executables; VSTest-style `dotnet test --filter` does not work. See the `tunit-testing` skill for full filter syntax.
3131

32+
**TUnit itself depends on this library.** `TUnit.Engine` is compiled against EnumerableAsyncProcessor v3, and the locally built assembly shadows the copy TUnit shipped with (same assembly identity), so removing or changing a public member that TUnit.Engine binds to crashes test discovery with `MissingMethodException` before any test runs. The exact signatures TUnit needs are pinned by `V3BinaryCompatibilityTests` and marked as binary-compat members in the source — do not remove them until TUnit ships a build compiled against v4.
33+
3234
CI (`.github/workflows/dotnet.yml`) runs the `EnumerableAsyncProcessor.Pipeline` project (a ModularPipelines app, `dotnet run -c Release` from that directory), which builds, tests, packs, and — on `main` — publishes to NuGet. Versioning comes from GitVersion (`GitVersion.yml` pins `next-version: 4.0.0`; keep that file present, its absence makes ModularPipelines generate a Mainline config that crashes on GitHub PR merge commits).
3335

3436
## Architecture
@@ -45,17 +47,17 @@ Processor classes vary along three axes, reflected in naming:
4547

4648
- **Input**: with items (`<TInput>`) vs. execution-count only (non-generic).
4749
- **Output**: `Result*`-prefixed classes (in `RunnableProcessors/ResultProcessors/`) return values via `IAsyncProcessor<TOutput>` (`GetResultsAsync()`, `GetResultsAsyncEnumerable()`, `GetEnumerableTasks()`); unprefixed classes are fire-and-await (`WaitAsync()`).
48-
- **Strategy**: `OneAtATime`, `Batch`, `Parallel`, `RateLimitedParallel`, `TimedRateLimitedParallel`.
50+
- **Strategy**: `OneAtATime`, `Batch`, `Parallel`, `TimedRateLimitedParallel`.
4951

50-
`RunnableProcessors/AsyncEnumerable/` holds parallel variants for `IAsyncEnumerable<T>` sources. File-name suffixes `_1`/`_2` distinguish generic arity (e.g. `BatchAsyncProcessor_1.cs` is `BatchAsyncProcessor<TInput>`).
52+
`RunnableProcessors/AsyncEnumerable/` holds the `IAsyncEnumerable<T>`-source variants (Parallel, OneAtATime, Batch). File-name suffixes `_1`/`_2` distinguish generic arity (e.g. `BatchAsyncProcessor_1.cs` is `BatchAsyncProcessor<TInput>`).
5153

5254
### Core mechanics (read these before changing behavior)
5355

5456
- **`ProcessorLifecycle.cs`**: owns start/cancel/dispose shared by both base-class hierarchies. `AbstractAsyncProcessorBase` (void) and `ResultAbstractAsyncProcessorBase` (results) cannot share an ancestor because they fan out to differently typed `TaskCompletionSource` lists, so both delegate to this class. Cancellation is registered in `Start`, not the constructor, so a pre-cancelled token can never fire on a partially built instance. `DisposeAsync` waits up to 30 seconds for in-flight tasks; sync `Dispose` cancels without blocking.
5557
- **TCS-per-item**: each item gets a `TaskCompletionSource`; `TaskWrapper.Process` never throws — it completes the item's TCS with the failure/cancellation instead, so one failed item cannot kill the run or leave awaiters hanging.
56-
- **`WorkerPool.cs`**: rate-limited processors run a fixed pool of worker loops claiming items via `Interlocked.Increment` (P `Task.Run` tasks total, not N throttled tasks + semaphore). `minimumIterationTime` is how timed rate limiting is implemented: each worker holds its slot for at least that duration per item.
58+
- **`WorkerPool.cs`**: rate-limited processors run a fixed pool of worker loops claiming items via `Interlocked.Increment` (P `Task.Run` tasks total, not N throttled tasks + semaphore). Timed rate limiting is a shared `TokenBucketRateLimiter` (`System.Threading.RateLimiting`): workers acquire a permit before starting each item, so `permitsPerWindow`/`window` bound the start rate independently of `maxConcurrency`.
5759
- **Multi-targeting**: `EnumerableExtensions.ToIAsyncEnumerable` uses `Task.WhenEach` on `NET9_0_OR_GREATER` and a completion-order-bucket fallback otherwise. The test project targets `net8.0` specifically to exercise the fallback path — don't drop that TFM.
5860

5961
### Disposal contract
6062

61-
All processors implement `IDisposable`/`IAsyncDisposable`; the README documents the patterns users rely on (`await using`, safe double/early disposal). The builder extension shortcuts on `IAsyncEnumerable<T>` dispose internally; processors returned from the builder pattern are the caller's responsibility. Preserve these semantics — there are dedicated regression tests (`DisposalRegressionTests`, `ExceptionFidelityTests`, `InputEnumerationRegressionTests`).
63+
All processors implement `IDisposable`/`IAsyncDisposable`; the README documents the patterns users rely on (`await using`, safe double/early disposal). `IAsyncEnumerableProcessor` implementations are single-use and additionally dispose their internal linked `CancellationTokenSource` when `ExecuteAsync` completes; `IAsyncProcessor` objects returned from the builder pattern are the caller's responsibility. Preserve these semantics — there are dedicated regression tests (`DisposalRegressionTests`, `ExceptionFidelityTests`, `InputEnumerationRegressionTests`).

EnumerableAsyncProcessor.Example/ProcessInParallelExample.cs

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -15,10 +15,10 @@ public static async Task RunExample()
1515
Console.WriteLine("ProcessInParallel Extension Examples");
1616
Console.WriteLine("====================================\n");
1717

18-
// Example 1: Simple parallel processing without transformation
19-
Console.WriteLine("Example 1: Simple parallel processing (no transformation needed!)");
18+
// Example 1: Simple parallel processing with an identity transformation
19+
Console.WriteLine("Example 1: Simple parallel processing");
2020
var asyncEnumerable1 = GenerateAsyncEnumerable(5);
21-
IEnumerable<int> results1 = await asyncEnumerable1.ProcessInParallel(); // <-- This is the simple extension!
21+
IEnumerable<int> results1 = await asyncEnumerable1.ProcessInParallel(item => Task.FromResult(item));
2222
Console.WriteLine($"Results: {string.Join(", ", results1)}");
2323

2424
// Example 2: Parallel processing with transformation

EnumerableAsyncProcessor.UnitTests/AsyncEnumerableParallelExtensionsTests.cs

Lines changed: 7 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -32,24 +32,13 @@ private static async IAsyncEnumerable<int> GenerateDelayedAsyncEnumerable(int co
3232
}
3333
}
3434

35-
[Test]
36-
public async Task ProcessInParallel_WithoutTransformation_ReturnsAllItems()
37-
{
38-
var asyncEnumerable = GenerateAsyncEnumerable(10);
39-
40-
var results = await asyncEnumerable.ProcessInParallel();
41-
42-
await Assert.That(results.Count()).IsEqualTo(10);
43-
await Assert.That(results.OrderBy(x => x)).IsEquivalentTo(Enumerable.Range(1, 10));
44-
}
45-
4635
[Test]
4736
public async Task ProcessInParallel_WithMaxConcurrency_ReturnsAllItems()
4837
{
4938
var asyncEnumerable = GenerateAsyncEnumerable(20);
50-
51-
var results = await asyncEnumerable.ProcessInParallel(5);
52-
39+
40+
var results = await asyncEnumerable.ProcessInParallel(item => Task.FromResult(item), 5);
41+
5342
await Assert.That(results.Count()).IsEqualTo(20);
5443
await Assert.That(results.OrderBy(x => x)).IsEquivalentTo(Enumerable.Range(1, 20));
5544
}
@@ -94,7 +83,7 @@ public async Task ProcessInParallel_WithCancellation_ThrowsOperationCanceledExce
9483
using var cts = new CancellationTokenSource();
9584
var asyncEnumerable = GenerateDelayedAsyncEnumerable(100, 50);
9685

97-
var task = asyncEnumerable.ProcessInParallel(cancellationToken: cts.Token);
86+
var task = asyncEnumerable.ProcessInParallel(item => Task.FromResult(item), cancellationToken: cts.Token);
9887

9988
// Cancel after a short delay
10089
cts.CancelAfter(100);
@@ -160,23 +149,10 @@ public async Task ProcessInParallel_WithMaxConcurrency_LimitsConcurrency()
160149
public async Task ProcessInParallel_EmptyEnumerable_ReturnsEmptyResult()
161150
{
162151
var asyncEnumerable = GenerateAsyncEnumerable(0);
163-
164-
var results = await asyncEnumerable.ProcessInParallel();
165-
166-
await Assert.That(results.Count()).IsEqualTo(0);
167-
}
168152

169-
[Test]
170-
public async Task ProcessInParallel_WithScheduleOnThreadPool_ProcessesAllItems()
171-
{
172-
var asyncEnumerable = GenerateAsyncEnumerable(10);
173-
174-
var results = await asyncEnumerable.ProcessInParallel(
175-
maxConcurrency: null,
176-
scheduleOnThreadPool: true);
177-
178-
await Assert.That(results.Count()).IsEqualTo(10);
179-
await Assert.That(results.OrderBy(x => x)).IsEquivalentTo(Enumerable.Range(1, 10));
153+
var results = await asyncEnumerable.ProcessInParallel(item => Task.FromResult(item));
154+
155+
await Assert.That(results.Count()).IsEqualTo(0);
180156
}
181157

182158
[Test]

EnumerableAsyncProcessor.UnitTests/DisposalRegressionTests.cs

Lines changed: 42 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,8 @@
11
using System;
2+
using System.Collections.Generic;
23
using System.Diagnostics;
34
using System.Linq;
5+
using System.Threading;
46
using System.Threading.Tasks;
57
using EnumerableAsyncProcessor.Extensions;
68

@@ -127,4 +129,44 @@ public async Task Disposal_Is_Idempotent_And_Safe_In_Any_Order()
127129

128130
await Assert.That(processor.GetEnumerableTasks().Count(x => x.IsCompletedSuccessfully)).IsEqualTo(5);
129131
}
132+
133+
[Test]
134+
public async Task AsyncEnumerable_Processor_Disposal_Is_Idempotent_After_Execution()
135+
{
136+
var processedCount = 0;
137+
138+
var processor = GenerateAsyncEnumerable(5)
139+
.ForEachAsync(_ =>
140+
{
141+
Interlocked.Increment(ref processedCount);
142+
return Task.CompletedTask;
143+
})
144+
.ProcessInParallel(maxConcurrency: 2);
145+
146+
await processor.ExecuteAsync();
147+
148+
// ExecuteAsync disposes internal resources on completion; explicit disposal stays safe.
149+
await processor.DisposeAsync();
150+
processor.Dispose();
151+
152+
await Assert.That(processedCount).IsEqualTo(5);
153+
}
154+
155+
[Test]
156+
public async Task AsyncEnumerable_Result_Processor_Supports_Await_Using_Without_Execution()
157+
{
158+
await using (GenerateAsyncEnumerable(3).SelectAsync(i => Task.FromResult(i)).ProcessInParallel(2))
159+
{
160+
// Never executed - disposal alone must not throw.
161+
}
162+
}
163+
164+
private static async IAsyncEnumerable<int> GenerateAsyncEnumerable(int count)
165+
{
166+
for (var i = 0; i < count; i++)
167+
{
168+
await Task.Yield();
169+
yield return i;
170+
}
171+
}
130172
}
Lines changed: 47 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,47 @@
1+
using System;
2+
using System.Collections.Generic;
3+
using System.Linq;
4+
using System.Threading;
5+
using System.Threading.Tasks;
6+
using EnumerableAsyncProcessor.Builders;
7+
using EnumerableAsyncProcessor.Extensions;
8+
9+
namespace EnumerableAsyncProcessor.UnitTests;
10+
11+
/// <summary>
12+
/// TUnit.Engine (the framework running this suite) is itself compiled against
13+
/// EnumerableAsyncProcessor, and the locally built assembly shadows the version TUnit shipped
14+
/// with because the assembly identity matches. If a member TUnit.Engine binds to disappears,
15+
/// test DISCOVERY crashes with MissingMethodException before any test runs. These are the
16+
/// exact signatures TUnit.Engine references - keep them until TUnit rebuilds against v4.
17+
/// </summary>
18+
public class V3BinaryCompatibilityTests
19+
{
20+
[Test]
21+
public async Task Members_Bound_By_TUnit_Engine_Exist_With_Exact_Signatures()
22+
{
23+
// ItemActionAsyncProcessorBuilder<TInput, TOutput>.ProcessInParallel(int)
24+
var builderMethod = typeof(ItemActionAsyncProcessorBuilder<,>).GetMethods()
25+
.SingleOrDefault(m => m.Name == "ProcessInParallel"
26+
&& m.GetParameters().Length == 1
27+
&& m.GetParameters()[0].ParameterType == typeof(int));
28+
await Assert.That(builderMethod).IsNotNull();
29+
30+
// AsyncEnumerableExtensions.ProcessInParallel<T>(IAsyncEnumerable<T>, CancellationToken)
31+
var extensionMethod = typeof(AsyncEnumerableExtensions).GetMethods()
32+
.SingleOrDefault(m => m.Name == "ProcessInParallel"
33+
&& m.GetGenericArguments().Length == 1
34+
&& m.GetParameters().Length == 2
35+
&& m.GetParameters()[1].ParameterType == typeof(CancellationToken));
36+
await Assert.That(extensionMethod).IsNotNull();
37+
38+
// TUnit.Engine also binds EnumerableExtensions.SelectAsync / SelectManyAsync and
39+
// IAsyncProcessor<T>.GetAwaiter; these exact-signature method groups stop compiling if they drift.
40+
Func<IEnumerable<int>, Func<int, Task<int>>, CancellationToken, ItemActionAsyncProcessorBuilder<int, int>> selectAsync =
41+
EnumerableExtensions.SelectAsync;
42+
Func<IEnumerable<int>, Func<int, IAsyncEnumerable<int>>, CancellationToken, IAsyncEnumerable<int>> selectManyAsync =
43+
EnumerableExtensions.SelectManyAsync;
44+
await Assert.That(selectAsync).IsNotNull();
45+
await Assert.That(selectManyAsync).IsNotNull();
46+
}
47+
}

EnumerableAsyncProcessor/Builders/ActionAsyncProcessorBuilder.cs

Lines changed: 10 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,7 @@
44

55
namespace EnumerableAsyncProcessor.Builders;
66

7-
public class ActionAsyncProcessorBuilder
7+
public sealed class ActionAsyncProcessorBuilder
88
{
99
private readonly int _count;
1010
private readonly Func<Task> _taskSelector;
@@ -53,6 +53,15 @@ public IAsyncProcessor ProcessInParallel(int? maxConcurrency = null, bool schedu
5353
{
5454
return new ParallelAsyncProcessor(_count, _taskSelector, _cancellationTokenSource, maxConcurrency, scheduleOnThreadPool).StartProcessing();
5555
}
56+
57+
/// <summary>
58+
/// Processes items in parallel with bounded concurrency. Binary-compatible with assemblies
59+
/// compiled against v3 (equivalent to <c>ProcessInParallel(maxConcurrency: n)</c>).
60+
/// </summary>
61+
public IAsyncProcessor ProcessInParallel(int maxConcurrency)
62+
{
63+
return ProcessInParallel((int?)maxConcurrency);
64+
}
5665

5766
public IAsyncProcessor ProcessOneAtATime()
5867
{

EnumerableAsyncProcessor/Builders/ActionAsyncProcessorBuilder_1.cs

Lines changed: 10 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,7 @@
44

55
namespace EnumerableAsyncProcessor.Builders;
66

7-
public class ActionAsyncProcessorBuilder<TOutput>
7+
public sealed class ActionAsyncProcessorBuilder<TOutput>
88
{
99
private readonly int _count;
1010
private readonly Func<Task<TOutput>> _taskSelector;
@@ -53,6 +53,15 @@ public IAsyncProcessor<TOutput> ProcessInParallel(int? maxConcurrency = null, bo
5353
{
5454
return new ResultParallelAsyncProcessor<TOutput>(_count, _taskSelector, _cancellationTokenSource, maxConcurrency, scheduleOnThreadPool).StartProcessing();
5555
}
56+
57+
/// <summary>
58+
/// Processes items in parallel with bounded concurrency. Binary-compatible with assemblies
59+
/// compiled against v3 (equivalent to <c>ProcessInParallel(maxConcurrency: n)</c>).
60+
/// </summary>
61+
public IAsyncProcessor<TOutput> ProcessInParallel(int maxConcurrency)
62+
{
63+
return ProcessInParallel((int?)maxConcurrency);
64+
}
5665

5766
public IAsyncProcessor<TOutput> ProcessOneAtATime()
5867
{

EnumerableAsyncProcessor/Builders/AsyncEnumerableActionAsyncProcessorBuilder_1.cs

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,9 @@
1-
using EnumerableAsyncProcessor.Extensions;
1+
using EnumerableAsyncProcessor.Interfaces;
22
using EnumerableAsyncProcessor.RunnableProcessors.AsyncEnumerable;
33

44
namespace EnumerableAsyncProcessor.Builders;
55

6-
public class AsyncEnumerableActionAsyncProcessorBuilder<TInput>
6+
public sealed class AsyncEnumerableActionAsyncProcessorBuilder<TInput>
77
{
88
private readonly IAsyncEnumerable<TInput> _items;
99
private readonly Func<TInput, Task> _taskSelector;

EnumerableAsyncProcessor/Builders/AsyncEnumerableActionAsyncProcessorBuilder_2.cs

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,9 @@
1-
using EnumerableAsyncProcessor.Extensions;
1+
using EnumerableAsyncProcessor.Interfaces;
22
using EnumerableAsyncProcessor.RunnableProcessors.AsyncEnumerable.ResultProcessors;
33

44
namespace EnumerableAsyncProcessor.Builders;
55

6-
public class AsyncEnumerableActionAsyncProcessorBuilder<TInput, TOutput>
6+
public sealed class AsyncEnumerableActionAsyncProcessorBuilder<TInput, TOutput>
77
{
88
private readonly IAsyncEnumerable<TInput> _items;
99
private readonly Func<TInput, Task<TOutput>> _taskSelector;

EnumerableAsyncProcessor/Builders/AsyncEnumerableAsyncProcessorBuilder.cs

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
namespace EnumerableAsyncProcessor.Builders;
22

3-
public class AsyncEnumerableAsyncProcessorBuilder<TInput>
3+
public sealed class AsyncEnumerableAsyncProcessorBuilder<TInput>
44
{
55
private readonly IAsyncEnumerable<TInput> _items;
66

0 commit comments

Comments
 (0)