Skip to content

Commit 9ea8230

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 928aff6 commit 9ea8230

14 files changed

Lines changed: 78 additions & 418 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: 6 additions & 40 deletions
Original file line numberDiff line numberDiff line change
@@ -29,52 +29,18 @@ public IAsyncProcessor ProcessInBatches(int batchSize)
2929
return new BatchAsyncProcessor(batchSize, _count, _taskSelector, _cancellationTokenSource).StartProcessing();
3030
}
3131

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

7137
/// <summary>
72-
/// Process tasks in parallel with specified concurrency limit.
38+
/// Process tasks in parallel, optionally limiting concurrency.
7339
/// </summary>
74-
/// <param name="maxConcurrency">Maximum concurrent operations.</param>
75-
/// <param name="scheduleOnThreadPool">If true, schedules tasks on thread pool to prevent blocking.</param>
40+
/// <param name="maxConcurrency">Maximum concurrent operations, or null for unbounded concurrency.</param>
41+
/// <param name="scheduleOnThreadPool">For unbounded processing, schedules tasks on the thread pool when true.</param>
7642
/// <returns>An async processor configured for parallel execution.</returns>
77-
public IAsyncProcessor ProcessInParallel(int? maxConcurrency, bool scheduleOnThreadPool)
43+
public IAsyncProcessor ProcessInParallel(int? maxConcurrency = null, bool scheduleOnThreadPool = false)
7844
{
7945
return new ParallelAsyncProcessor(_count, _taskSelector, _cancellationTokenSource, maxConcurrency, scheduleOnThreadPool).StartProcessing();
8046
}

‎EnumerableAsyncProcessor/Builders/ActionAsyncProcessorBuilder_1.cs‎

Lines changed: 6 additions & 40 deletions
Original file line numberDiff line numberDiff line change
@@ -29,52 +29,18 @@ public IAsyncProcessor<TOutput> ProcessInBatches(int batchSize)
2929
return new ResultBatchAsyncProcessor<TOutput>(batchSize, _count, _taskSelector, _cancellationTokenSource).StartProcessing();
3030
}
3131

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

7137
/// <summary>
72-
/// Process tasks in parallel with specified concurrency limit and return results.
38+
/// Process tasks in parallel, optionally limiting concurrency, and return results.
7339
/// </summary>
74-
/// <param name="maxConcurrency">Maximum concurrent operations.</param>
75-
/// <param name="scheduleOnThreadPool">If true, schedules tasks on thread pool to prevent blocking.</param>
40+
/// <param name="maxConcurrency">Maximum concurrent operations, or null for unbounded concurrency.</param>
41+
/// <param name="scheduleOnThreadPool">For unbounded processing, schedules tasks on the thread pool when true.</param>
7642
/// <returns>An async processor configured for parallel execution that returns results.</returns>
77-
public IAsyncProcessor<TOutput> ProcessInParallel(int? maxConcurrency, bool scheduleOnThreadPool)
43+
public IAsyncProcessor<TOutput> ProcessInParallel(int? maxConcurrency = null, bool scheduleOnThreadPool = false)
7844
{
7945
return new ResultParallelAsyncProcessor<TOutput>(_count, _taskSelector, _cancellationTokenSource, maxConcurrency, scheduleOnThreadPool).StartProcessing();
8046
}

‎EnumerableAsyncProcessor/Builders/AsyncEnumerableActionAsyncProcessorBuilder_1.cs‎

Lines changed: 4 additions & 43 deletions
Original file line numberDiff line numberDiff line change
@@ -30,51 +30,12 @@ public AsyncEnumerableActionAsyncProcessorBuilder(
3030
}
3131

3232
/// <summary>
33-
/// Process items in parallel without concurrency limits.
33+
/// Process items in parallel, optionally limiting concurrency.
3434
/// </summary>
35+
/// <param name="maxConcurrency">Maximum concurrent operations, or null for unbounded concurrency.</param>
36+
/// <param name="scheduleOnThreadPool">For unbounded processing, schedules tasks on the thread pool when true.</param>
3537
/// <returns>An async processor configured for parallel execution.</returns>
36-
public IAsyncEnumerableProcessor ProcessInParallel()
37-
{
38-
return ProcessInParallel(null, false);
39-
}
40-
41-
/// <summary>
42-
/// Process items in parallel without concurrency limits.
43-
/// </summary>
44-
/// <param name="scheduleOnThreadPool">If true, schedules tasks on thread pool to prevent blocking. Default is false for maximum performance.</param>
45-
/// <returns>An async processor configured for parallel execution.</returns>
46-
public IAsyncEnumerableProcessor ProcessInParallel(bool scheduleOnThreadPool)
47-
{
48-
return ProcessInParallel(null, scheduleOnThreadPool);
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((int?)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-
/// <returns>An async processor configured for parallel execution.</returns>
66-
public IAsyncEnumerableProcessor ProcessInParallel(int? maxConcurrency)
67-
{
68-
return ProcessInParallel(maxConcurrency, false);
69-
}
70-
71-
/// <summary>
72-
/// Process items in parallel with specified concurrency limit.
73-
/// </summary>
74-
/// <param name="maxConcurrency">Maximum concurrent operations.</param>
75-
/// <param name="scheduleOnThreadPool">If true, schedules tasks on thread pool to prevent blocking.</param>
76-
/// <returns>An async processor configured for parallel execution.</returns>
77-
public IAsyncEnumerableProcessor ProcessInParallel(int? maxConcurrency, bool scheduleOnThreadPool)
38+
public IAsyncEnumerableProcessor ProcessInParallel(int? maxConcurrency = null, bool scheduleOnThreadPool = false)
7839
{
7940
return new AsyncEnumerableParallelProcessor<TInput>(
8041
_items, _taskSelector, maxConcurrency, scheduleOnThreadPool, _cancellationTokenSource);

‎EnumerableAsyncProcessor/Builders/AsyncEnumerableActionAsyncProcessorBuilder_2.cs‎

Lines changed: 4 additions & 43 deletions
Original file line numberDiff line numberDiff line change
@@ -30,51 +30,12 @@ public AsyncEnumerableActionAsyncProcessorBuilder(
3030
}
3131

3232
/// <summary>
33-
/// Process items in parallel without concurrency limits and return results.
33+
/// Process items in parallel, optionally limiting concurrency, and return results.
3434
/// </summary>
35+
/// <param name="maxConcurrency">Maximum concurrent operations, or null for unbounded concurrency.</param>
36+
/// <param name="scheduleOnThreadPool">For unbounded processing, schedules tasks on the thread pool when true.</param>
3537
/// <returns>An async processor configured for parallel execution.</returns>
36-
public IAsyncEnumerableProcessor<TOutput> ProcessInParallel()
37-
{
38-
return ProcessInParallel(null, false);
39-
}
40-
41-
/// <summary>
42-
/// Process items in parallel without concurrency limits and return results.
43-
/// </summary>
44-
/// <param name="scheduleOnThreadPool">If true, schedules tasks on thread pool to prevent blocking. Default is false for maximum performance.</param>
45-
/// <returns>An async processor configured for parallel execution.</returns>
46-
public IAsyncEnumerableProcessor<TOutput> ProcessInParallel(bool scheduleOnThreadPool)
47-
{
48-
return ProcessInParallel(null, scheduleOnThreadPool);
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((int?)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-
/// <returns>An async processor configured for parallel execution.</returns>
66-
public IAsyncEnumerableProcessor<TOutput> ProcessInParallel(int? maxConcurrency)
67-
{
68-
return ProcessInParallel(maxConcurrency, false);
69-
}
70-
71-
/// <summary>
72-
/// Process items in parallel with specified concurrency limit and return results.
73-
/// </summary>
74-
/// <param name="maxConcurrency">Maximum concurrent operations.</param>
75-
/// <param name="scheduleOnThreadPool">If true, schedules tasks on thread pool to prevent blocking.</param>
76-
/// <returns>An async processor configured for parallel execution.</returns>
77-
public IAsyncEnumerableProcessor<TOutput> ProcessInParallel(int? maxConcurrency, bool scheduleOnThreadPool)
38+
public IAsyncEnumerableProcessor<TOutput> ProcessInParallel(int? maxConcurrency = null, bool scheduleOnThreadPool = false)
7839
{
7940
return new ResultAsyncEnumerableParallelProcessor<TInput, TOutput>(
8041
_items, _taskSelector, maxConcurrency, scheduleOnThreadPool, _cancellationTokenSource);

‎EnumerableAsyncProcessor/Builders/ItemActionAsyncProcessorBuilder_1.cs‎

Lines changed: 6 additions & 41 deletions
Original file line numberDiff line numberDiff line change
@@ -30,54 +30,19 @@ public IAsyncProcessor ProcessInBatches(int batchSize)
3030
.StartProcessing();
3131
}
3232

33-
public IAsyncProcessor ProcessInParallel(int levelOfParallelism)
33+
public IAsyncProcessor ProcessInParallel(int maxConcurrency, TimeSpan timeSpan)
3434
{
35-
return new RateLimitedParallelAsyncProcessor<TInput>(_items, _taskSelector, levelOfParallelism, _cancellationTokenSource)
35+
return new TimedRateLimitedParallelAsyncProcessor<TInput>(_items, _taskSelector, maxConcurrency, timeSpan, _cancellationTokenSource)
3636
.StartProcessing();
3737
}
3838

39-
public IAsyncProcessor ProcessInParallel(int levelOfParallelism, TimeSpan timeSpan)
40-
{
41-
return new TimedRateLimitedParallelAsyncProcessor<TInput>(_items, _taskSelector, levelOfParallelism, timeSpan, _cancellationTokenSource)
42-
.StartProcessing();
43-
}
44-
45-
/// <summary>
46-
/// Process items in parallel without concurrency limits.
47-
/// </summary>
48-
/// <returns>An async processor configured for parallel execution.</returns>
49-
public IAsyncProcessor ProcessInParallel()
50-
{
51-
return ProcessInParallel(null, false);
52-
}
53-
54-
/// <summary>
55-
/// Process items in parallel without concurrency limits.
56-
/// </summary>
57-
/// <param name="scheduleOnThreadPool">If true, schedules tasks on thread pool to prevent blocking. Default is false for maximum performance.</param>
58-
/// <returns>An async processor configured for parallel execution.</returns>
59-
public IAsyncProcessor ProcessInParallel(bool scheduleOnThreadPool)
60-
{
61-
return ProcessInParallel(null, scheduleOnThreadPool);
62-
}
63-
64-
/// <summary>
65-
/// Process items in parallel with specified concurrency limit.
66-
/// </summary>
67-
/// <param name="maxConcurrency">Maximum concurrent operations.</param>
68-
/// <returns>An async processor configured for parallel execution.</returns>
69-
public IAsyncProcessor ProcessInParallel(int? maxConcurrency)
70-
{
71-
return ProcessInParallel(maxConcurrency, false);
72-
}
73-
7439
/// <summary>
75-
/// Process items in parallel with specified concurrency limit.
40+
/// Process items in parallel, optionally limiting concurrency.
7641
/// </summary>
77-
/// <param name="maxConcurrency">Maximum concurrent operations.</param>
78-
/// <param name="scheduleOnThreadPool">If true, schedules tasks on thread pool to prevent blocking.</param>
42+
/// <param name="maxConcurrency">Maximum concurrent operations, or null for unbounded concurrency.</param>
43+
/// <param name="scheduleOnThreadPool">For unbounded processing, schedules tasks on the thread pool when true.</param>
7944
/// <returns>An async processor configured for parallel execution.</returns>
80-
public IAsyncProcessor ProcessInParallel(int? maxConcurrency, bool scheduleOnThreadPool)
45+
public IAsyncProcessor ProcessInParallel(int? maxConcurrency = null, bool scheduleOnThreadPool = false)
8146
{
8247
return new ParallelAsyncProcessor<TInput>(_items, _taskSelector, _cancellationTokenSource, maxConcurrency, scheduleOnThreadPool)
8348
.StartProcessing();

0 commit comments

Comments
 (0)