Skip to content

Commit 3feffe0

Browse files
thomhurstthomhurst
committed
refactor!: consolidate parallel API
Route positional and named concurrency limits through ParallelAsyncProcessor and remove the duplicate non-timed rate-limited processor types. Keep timed rate limiting separate and document one optional-limit API.
1 parent adde5f0 commit 3feffe0

14 files changed

Lines changed: 84 additions & 426 deletions

EnumerableAsyncProcessor.UnitTests/RateLimitedParallelAsyncProcessorTests.cs renamed to EnumerableAsyncProcessor.UnitTests/BoundedParallelAsyncProcessorTests.cs

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,7 @@
88

99
namespace EnumerableAsyncProcessor.UnitTests;
1010

11-
public class RateLimitedParallelAsyncProcessorTests
11+
public class BoundedParallelAsyncProcessorTests
1212
{
1313
[MatrixDataSource]
1414
[Test, Repeat(5)]
@@ -156,4 +156,4 @@ public async Task When_Less_Tasks_Remaining_Than_Parallel_Limit_Then_Tasks_Remai
156156
await Assert.That(processor.GetEnumerableTasks().Count(x => x.Status == TaskStatus.RanToCompletion)).IsEqualTo(47);
157157
await Assert.That(processor.GetEnumerableTasks().Count(x => x.Status == TaskStatus.WaitingForActivation)).IsEqualTo(3);
158158
}
159-
}
159+
}

‎EnumerableAsyncProcessor.UnitTests/ValidationRegressionTests.cs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -31,7 +31,7 @@ await AssertThrows<ArgumentOutOfRangeException>(() =>
3131
}
3232

3333
[Test]
34-
public async Task Zero_Parallelism_Throws_For_Void_And_Result_RateLimited_Processors()
34+
public async Task Zero_MaxConcurrency_Throws_For_Positional_Void_And_Result_Calls()
3535
{
3636
await AssertThrows<ArgumentOutOfRangeException>(() =>
3737
new[] { 1 }.ForEachAsync(_ => Task.CompletedTask).ProcessInParallel(0));

‎EnumerableAsyncProcessor.UnitTests/WorkerPoolBehaviourTests.cs‎

Lines changed: 21 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -4,17 +4,32 @@
44
using System.Threading;
55
using System.Threading.Tasks;
66
using EnumerableAsyncProcessor.Extensions;
7+
using EnumerableAsyncProcessor.RunnableProcessors;
78

89
namespace EnumerableAsyncProcessor.UnitTests;
910

1011
/// <summary>
11-
/// Guards the worker-pool execution model used by the rate-limited, timed and
12-
/// throttled-parallel processors: the concurrency limit must hold, every item must be
12+
/// Guards the worker-pool execution model used by the bounded parallel and timed
13+
/// rate-limited processors: the concurrency limit must hold, every item must be
1314
/// processed exactly once, oversized limits must not break anything, and cancellation
1415
/// must promptly cancel the unprocessed remainder.
1516
/// </summary>
1617
public class WorkerPoolBehaviourTests
1718
{
19+
[Test]
20+
public async Task Positional_And_Named_Concurrency_Limits_Use_The_Same_Processor()
21+
{
22+
await using var positional = new[] { 1 }
23+
.ForEachAsync(_ => Task.CompletedTask)
24+
.ProcessInParallel(1);
25+
await using var named = new[] { 1 }
26+
.ForEachAsync(_ => Task.CompletedTask)
27+
.ProcessInParallel(maxConcurrency: 1);
28+
29+
await Assert.That(positional.GetType()).IsEqualTo(typeof(ParallelAsyncProcessor<int>));
30+
await Assert.That(named.GetType()).IsEqualTo(typeof(ParallelAsyncProcessor<int>));
31+
}
32+
1833
[Test, Repeat(3)]
1934
public async Task MaxConcurrency_Path_Obeys_The_Limit_And_Processes_Everything(CancellationToken cancellationToken)
2035
{
@@ -47,7 +62,7 @@ public async Task MaxConcurrency_Path_Obeys_The_Limit_And_Processes_Everything(C
4762
}
4863

4964
[Test]
50-
public async Task Every_Item_Is_Processed_Exactly_Once_Under_Rate_Limiting()
65+
public async Task Every_Item_Is_Processed_Exactly_Once_With_Bounded_Concurrency()
5166
{
5267
const int itemCount = 500;
5368

@@ -70,17 +85,17 @@ public async Task Every_Item_Is_Processed_Exactly_Once_Under_Rate_Limiting()
7085
[Test]
7186
public async Task Parallelism_Limit_Larger_Than_Item_Count_Completes_Normally()
7287
{
73-
await using var rateLimited = Enumerable.Range(0, 5).ToList()
88+
await using var bounded = Enumerable.Range(0, 5).ToList()
7489
.ForEachAsync(_ => Task.CompletedTask)
7590
.ProcessInParallel(100);
76-
await rateLimited.WaitAsync();
91+
await bounded.WaitAsync();
7792

7893
await using var throttled = Enumerable.Range(0, 5).ToList()
7994
.SelectAsync(i => Task.FromResult(i))
8095
.ProcessInParallel(maxConcurrency: 100);
8196
var results = await throttled.GetResultsAsync();
8297

83-
await Assert.That(rateLimited.GetEnumerableTasks().Count(x => x.IsCompletedSuccessfully)).IsEqualTo(5);
98+
await Assert.That(bounded.GetEnumerableTasks().Count(x => x.IsCompletedSuccessfully)).IsEqualTo(5);
8499
await Assert.That(results.Length).IsEqualTo(5);
85100
}
86101

‎EnumerableAsyncProcessor/Builders/ActionAsyncProcessorBuilder.cs‎

Lines changed: 7 additions & 41 deletions
Original file line numberDiff line numberDiff line change
@@ -22,52 +22,18 @@ public IAsyncProcessor ProcessInBatches(int batchSize)
2222
return new BatchAsyncProcessor(batchSize, _count, _taskSelector, _cancellationTokenSource).StartProcessing();
2323
}
2424

25-
public IAsyncProcessor ProcessInParallel(int levelOfParallelism)
25+
public IAsyncProcessor ProcessInParallel(int maxConcurrency, TimeSpan timeSpan)
2626
{
27-
return new RateLimitedParallelAsyncProcessor(_count, _taskSelector, levelOfParallelism, _cancellationTokenSource).StartProcessing();
28-
}
29-
30-
public IAsyncProcessor ProcessInParallel(int levelOfParallelism, TimeSpan timeSpan)
31-
{
32-
return new TimedRateLimitedParallelAsyncProcessor(_count, _taskSelector, levelOfParallelism, timeSpan, _cancellationTokenSource).StartProcessing();
33-
}
34-
35-
/// <summary>
36-
/// Process tasks in parallel without concurrency limits.
37-
/// </summary>
38-
/// <returns>An async processor configured for parallel execution.</returns>
39-
public IAsyncProcessor ProcessInParallel()
40-
{
41-
return ProcessInParallel(null, false);
42-
}
43-
44-
/// <summary>
45-
/// Process tasks in parallel without concurrency limits.
46-
/// </summary>
47-
/// <param name="scheduleOnThreadPool">If true, schedules tasks on thread pool to prevent blocking. Default is false for maximum performance.</param>
48-
/// <returns>An async processor configured for parallel execution.</returns>
49-
public IAsyncProcessor ProcessInParallel(bool scheduleOnThreadPool)
50-
{
51-
return ProcessInParallel(null, scheduleOnThreadPool);
52-
}
53-
54-
/// <summary>
55-
/// Process tasks in parallel with specified concurrency limit.
56-
/// </summary>
57-
/// <param name="maxConcurrency">Maximum concurrent operations.</param>
58-
/// <returns>An async processor configured for parallel execution.</returns>
59-
public IAsyncProcessor ProcessInParallel(int? maxConcurrency)
60-
{
61-
return ProcessInParallel(maxConcurrency, false);
27+
return new TimedRateLimitedParallelAsyncProcessor(_count, _taskSelector, maxConcurrency, timeSpan, _cancellationTokenSource).StartProcessing();
6228
}
6329

6430
/// <summary>
65-
/// Process tasks in parallel with specified concurrency limit.
31+
/// Process tasks in parallel, optionally limiting concurrency.
6632
/// </summary>
67-
/// <param name="maxConcurrency">Maximum concurrent operations.</param>
68-
/// <param name="scheduleOnThreadPool">If true, schedules tasks on thread pool to prevent blocking.</param>
33+
/// <param name="maxConcurrency">Maximum concurrent operations, or null for unbounded concurrency.</param>
34+
/// <param name="scheduleOnThreadPool">For unbounded processing, schedules tasks on the thread pool when true.</param>
6935
/// <returns>An async processor configured for parallel execution.</returns>
70-
public IAsyncProcessor ProcessInParallel(int? maxConcurrency, bool scheduleOnThreadPool)
36+
public IAsyncProcessor ProcessInParallel(int? maxConcurrency = null, bool scheduleOnThreadPool = false)
7137
{
7238
return new ParallelAsyncProcessor(_count, _taskSelector, _cancellationTokenSource, maxConcurrency, scheduleOnThreadPool).StartProcessing();
7339
}
@@ -77,4 +43,4 @@ public IAsyncProcessor ProcessOneAtATime()
7743
return new OneAtATimeAsyncProcessor(_count, _taskSelector, _cancellationTokenSource).StartProcessing();
7844
}
7945

80-
}
46+
}

‎EnumerableAsyncProcessor/Builders/ActionAsyncProcessorBuilder_1.cs‎

Lines changed: 7 additions & 41 deletions
Original file line numberDiff line numberDiff line change
@@ -22,52 +22,18 @@ public IAsyncProcessor<TOutput> ProcessInBatches(int batchSize)
2222
return new ResultBatchAsyncProcessor<TOutput>(batchSize, _count, _taskSelector, _cancellationTokenSource).StartProcessing();
2323
}
2424

25-
public IAsyncProcessor<TOutput> ProcessInParallel(int levelOfParallelism)
25+
public IAsyncProcessor<TOutput> ProcessInParallel(int maxConcurrency, TimeSpan timeSpan)
2626
{
27-
return new ResultRateLimitedParallelAsyncProcessor<TOutput>(_count, _taskSelector, levelOfParallelism, _cancellationTokenSource).StartProcessing();
28-
}
29-
30-
public IAsyncProcessor<TOutput> ProcessInParallel(int levelOfParallelism, TimeSpan timeSpan)
31-
{
32-
return new ResultTimedRateLimitedParallelAsyncProcessor<TOutput>(_count, _taskSelector, levelOfParallelism, timeSpan, _cancellationTokenSource).StartProcessing();
33-
}
34-
35-
/// <summary>
36-
/// Process tasks in parallel without concurrency limits and return results.
37-
/// </summary>
38-
/// <returns>An async processor configured for parallel execution that returns results.</returns>
39-
public IAsyncProcessor<TOutput> ProcessInParallel()
40-
{
41-
return ProcessInParallel(null, false);
42-
}
43-
44-
/// <summary>
45-
/// Process tasks in parallel without concurrency limits and return results.
46-
/// </summary>
47-
/// <param name="scheduleOnThreadPool">If true, schedules tasks on thread pool to prevent blocking. Default is false for maximum performance.</param>
48-
/// <returns>An async processor configured for parallel execution that returns results.</returns>
49-
public IAsyncProcessor<TOutput> ProcessInParallel(bool scheduleOnThreadPool)
50-
{
51-
return ProcessInParallel(null, scheduleOnThreadPool);
52-
}
53-
54-
/// <summary>
55-
/// Process tasks in parallel with specified concurrency limit and return results.
56-
/// </summary>
57-
/// <param name="maxConcurrency">Maximum concurrent operations.</param>
58-
/// <returns>An async processor configured for parallel execution that returns results.</returns>
59-
public IAsyncProcessor<TOutput> ProcessInParallel(int? maxConcurrency)
60-
{
61-
return ProcessInParallel(maxConcurrency, false);
27+
return new ResultTimedRateLimitedParallelAsyncProcessor<TOutput>(_count, _taskSelector, maxConcurrency, timeSpan, _cancellationTokenSource).StartProcessing();
6228
}
6329

6430
/// <summary>
65-
/// Process tasks in parallel with specified concurrency limit and return results.
31+
/// Process tasks in parallel, optionally limiting concurrency, and return results.
6632
/// </summary>
67-
/// <param name="maxConcurrency">Maximum concurrent operations.</param>
68-
/// <param name="scheduleOnThreadPool">If true, schedules tasks on thread pool to prevent blocking.</param>
33+
/// <param name="maxConcurrency">Maximum concurrent operations, or null for unbounded concurrency.</param>
34+
/// <param name="scheduleOnThreadPool">For unbounded processing, schedules tasks on the thread pool when true.</param>
6935
/// <returns>An async processor configured for parallel execution that returns results.</returns>
70-
public IAsyncProcessor<TOutput> ProcessInParallel(int? maxConcurrency, bool scheduleOnThreadPool)
36+
public IAsyncProcessor<TOutput> ProcessInParallel(int? maxConcurrency = null, bool scheduleOnThreadPool = false)
7137
{
7238
return new ResultParallelAsyncProcessor<TOutput>(_count, _taskSelector, _cancellationTokenSource, maxConcurrency, scheduleOnThreadPool).StartProcessing();
7339
}
@@ -77,4 +43,4 @@ public IAsyncProcessor<TOutput> ProcessOneAtATime()
7743
return new ResultOneAtATimeAsyncProcessor<TOutput>(_count, _taskSelector, _cancellationTokenSource).StartProcessing();
7844
}
7945

80-
}
46+
}

‎EnumerableAsyncProcessor/Builders/AsyncEnumerableActionAsyncProcessorBuilder_1.cs‎

Lines changed: 5 additions & 45 deletions
Original file line numberDiff line numberDiff line change
@@ -20,57 +20,17 @@ public AsyncEnumerableActionAsyncProcessorBuilder(
2020
}
2121

2222
/// <summary>
23-
/// Process items in parallel without concurrency limits.
23+
/// Process items in parallel, optionally limiting concurrency.
2424
/// </summary>
25+
/// <param name="maxConcurrency">Maximum concurrent operations, or null for unbounded concurrency.</param>
26+
/// <param name="scheduleOnThreadPool">For unbounded processing, schedules tasks on the thread pool when true.</param>
2527
/// <returns>An async processor configured for parallel execution.</returns>
26-
public IAsyncEnumerableProcessor ProcessInParallel()
27-
{
28-
return ProcessInParallel(null, false);
29-
}
30-
31-
/// <summary>
32-
/// Process items in parallel without concurrency limits.
33-
/// </summary>
34-
/// <param name="scheduleOnThreadPool">If true, schedules tasks on thread pool to prevent blocking. Default is false for maximum performance.</param>
35-
/// <returns>An async processor configured for parallel execution.</returns>
36-
public IAsyncEnumerableProcessor ProcessInParallel(bool scheduleOnThreadPool)
37-
{
38-
return ProcessInParallel(null, scheduleOnThreadPool);
39-
}
40-
41-
/// <summary>
42-
/// Process items in parallel with specified concurrency limit.
43-
/// </summary>
44-
/// <param name="maxConcurrency">Maximum concurrent operations.</param>
45-
/// <returns>An async processor configured for parallel execution.</returns>
46-
public IAsyncEnumerableProcessor ProcessInParallel(int maxConcurrency)
47-
{
48-
return ProcessInParallel((int?)maxConcurrency, false);
49-
}
50-
51-
/// <summary>
52-
/// Process items in parallel with specified concurrency limit.
53-
/// </summary>
54-
/// <param name="maxConcurrency">Maximum concurrent operations.</param>
55-
/// <returns>An async processor configured for parallel execution.</returns>
56-
public IAsyncEnumerableProcessor ProcessInParallel(int? maxConcurrency)
57-
{
58-
return ProcessInParallel(maxConcurrency, false);
59-
}
60-
61-
/// <summary>
62-
/// Process items in parallel with specified concurrency limit.
63-
/// </summary>
64-
/// <param name="maxConcurrency">Maximum concurrent operations.</param>
65-
/// <param name="scheduleOnThreadPool">If true, schedules tasks on thread pool to prevent blocking.</param>
66-
/// <returns>An async processor configured for parallel execution.</returns>
67-
public IAsyncEnumerableProcessor ProcessInParallel(int? maxConcurrency, bool scheduleOnThreadPool)
28+
public IAsyncEnumerableProcessor ProcessInParallel(int? maxConcurrency = null, bool scheduleOnThreadPool = false)
6829
{
6930
return new AsyncEnumerableParallelProcessor<TInput>(
7031
_items, _taskSelector, maxConcurrency, scheduleOnThreadPool, _cancellationTokenSource);
7132
}
7233

73-
7434
/// <summary>
7535
/// Process items one at a time (sequential processing).
7636
/// </summary>
@@ -91,4 +51,4 @@ public IAsyncEnumerableProcessor ProcessInBatches(int batchSize)
9151
_items, _taskSelector, batchSize, _cancellationTokenSource);
9252
}
9353

94-
}
54+
}

‎EnumerableAsyncProcessor/Builders/AsyncEnumerableActionAsyncProcessorBuilder_2.cs‎

Lines changed: 5 additions & 45 deletions
Original file line numberDiff line numberDiff line change
@@ -20,57 +20,17 @@ public AsyncEnumerableActionAsyncProcessorBuilder(
2020
}
2121

2222
/// <summary>
23-
/// Process items in parallel without concurrency limits and return results.
23+
/// Process items in parallel, optionally limiting concurrency, and return results.
2424
/// </summary>
25+
/// <param name="maxConcurrency">Maximum concurrent operations, or null for unbounded concurrency.</param>
26+
/// <param name="scheduleOnThreadPool">For unbounded processing, schedules tasks on the thread pool when true.</param>
2527
/// <returns>An async processor configured for parallel execution.</returns>
26-
public IAsyncEnumerableProcessor<TOutput> ProcessInParallel()
27-
{
28-
return ProcessInParallel(null, false);
29-
}
30-
31-
/// <summary>
32-
/// Process items in parallel without concurrency limits and return results.
33-
/// </summary>
34-
/// <param name="scheduleOnThreadPool">If true, schedules tasks on thread pool to prevent blocking. Default is false for maximum performance.</param>
35-
/// <returns>An async processor configured for parallel execution.</returns>
36-
public IAsyncEnumerableProcessor<TOutput> ProcessInParallel(bool scheduleOnThreadPool)
37-
{
38-
return ProcessInParallel(null, scheduleOnThreadPool);
39-
}
40-
41-
/// <summary>
42-
/// Process items in parallel with specified concurrency limit and return results.
43-
/// </summary>
44-
/// <param name="maxConcurrency">Maximum concurrent operations.</param>
45-
/// <returns>An async processor configured for parallel execution.</returns>
46-
public IAsyncEnumerableProcessor<TOutput> ProcessInParallel(int maxConcurrency)
47-
{
48-
return ProcessInParallel((int?)maxConcurrency, false);
49-
}
50-
51-
/// <summary>
52-
/// Process items in parallel with specified concurrency limit and return results.
53-
/// </summary>
54-
/// <param name="maxConcurrency">Maximum concurrent operations.</param>
55-
/// <returns>An async processor configured for parallel execution.</returns>
56-
public IAsyncEnumerableProcessor<TOutput> ProcessInParallel(int? maxConcurrency)
57-
{
58-
return ProcessInParallel(maxConcurrency, false);
59-
}
60-
61-
/// <summary>
62-
/// Process items in parallel with specified concurrency limit and return results.
63-
/// </summary>
64-
/// <param name="maxConcurrency">Maximum concurrent operations.</param>
65-
/// <param name="scheduleOnThreadPool">If true, schedules tasks on thread pool to prevent blocking.</param>
66-
/// <returns>An async processor configured for parallel execution.</returns>
67-
public IAsyncEnumerableProcessor<TOutput> ProcessInParallel(int? maxConcurrency, bool scheduleOnThreadPool)
28+
public IAsyncEnumerableProcessor<TOutput> ProcessInParallel(int? maxConcurrency = null, bool scheduleOnThreadPool = false)
6829
{
6930
return new ResultAsyncEnumerableParallelProcessor<TInput, TOutput>(
7031
_items, _taskSelector, maxConcurrency, scheduleOnThreadPool, _cancellationTokenSource);
7132
}
7233

73-
7434
/// <summary>
7535
/// Process items one at a time and return results in order.
7636
/// </summary>
@@ -91,4 +51,4 @@ public IAsyncEnumerableProcessor<TOutput> ProcessInBatches(int batchSize)
9151
_items, _taskSelector, batchSize, _cancellationTokenSource);
9252
}
9353

94-
}
54+
}

0 commit comments

Comments
 (0)