Skip to content

Consolidate ProcessInParallel API #352

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
thomhurst merged 1 commit into from
Jul 21, 2026
Merged

Consolidate ProcessInParallel API #352

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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: 2 additions & 2 deletions ...RateLimitedParallelAsyncProcessorTests.cs → ...sts/BoundedParallelAsyncProcessorTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,7 @@

namespace EnumerableAsyncProcessor.UnitTests;

public class RateLimitedParallelAsyncProcessorTests
public class BoundedParallelAsyncProcessorTests
{
[MatrixDataSource]
[Test, Repeat(5)]
Expand Down Expand Up @@ -156,4 +156,4 @@ public async Task When_Less_Tasks_Remaining_Than_Parallel_Limit_Then_Tasks_Remai
await Assert.That(processor.GetEnumerableTasks().Count(x => x.Status == TaskStatus.RanToCompletion)).IsEqualTo(47);
await Assert.That(processor.GetEnumerableTasks().Count(x => x.Status == TaskStatus.WaitingForActivation)).IsEqualTo(3);
}
}
}
2 changes: 1 addition & 1 deletion EnumerableAsyncProcessor.UnitTests/ValidationRegressionTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,7 @@ await AssertThrows<ArgumentOutOfRangeException>(() =>
}

[Test]
public async Task Zero_Parallelism_Throws_For_Void_And_Result_RateLimited_Processors()
public async Task Zero_MaxConcurrency_Throws_For_Positional_Void_And_Result_Calls()
{
await AssertThrows<ArgumentOutOfRangeException>(() =>
new[] { 1 }.ForEachAsync(_ => Task.CompletedTask).ProcessInParallel(0));
Expand Down
27 changes: 21 additions & 6 deletions EnumerableAsyncProcessor.UnitTests/WorkerPoolBehaviourTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,17 +4,32 @@
using System.Threading;
using System.Threading.Tasks;
using EnumerableAsyncProcessor.Extensions;
using EnumerableAsyncProcessor.RunnableProcessors;

namespace EnumerableAsyncProcessor.UnitTests;

/// <summary>
/// Guards the worker-pool execution model used by the rate-limited, timed and
/// throttled-parallel processors: the concurrency limit must hold, every item must be
/// Guards the worker-pool execution model used by the bounded parallel and timed
/// rate-limited processors: the concurrency limit must hold, every item must be
/// processed exactly once, oversized limits must not break anything, and cancellation
/// must promptly cancel the unprocessed remainder.
/// </summary>
public class WorkerPoolBehaviourTests
{
[Test]
public async Task Positional_And_Named_Concurrency_Limits_Use_The_Same_Processor()
{
await using var positional = new[] { 1 }
.ForEachAsync(_ => Task.CompletedTask)
.ProcessInParallel(1);
await using var named = new[] { 1 }
.ForEachAsync(_ => Task.CompletedTask)
.ProcessInParallel(maxConcurrency: 1);

await Assert.That(positional.GetType()).IsEqualTo(typeof(ParallelAsyncProcessor<int>));
await Assert.That(named.GetType()).IsEqualTo(typeof(ParallelAsyncProcessor<int>));
}

[Test, Repeat(3)]
public async Task MaxConcurrency_Path_Obeys_The_Limit_And_Processes_Everything(CancellationToken cancellationToken)
{
Expand Down Expand Up @@ -47,7 +62,7 @@ public async Task MaxConcurrency_Path_Obeys_The_Limit_And_Processes_Everything(C
}

[Test]
public async Task Every_Item_Is_Processed_Exactly_Once_Under_Rate_Limiting()
public async Task Every_Item_Is_Processed_Exactly_Once_With_Bounded_Concurrency()
{
const int itemCount = 500;

Expand All @@ -70,17 +85,17 @@ public async Task Every_Item_Is_Processed_Exactly_Once_Under_Rate_Limiting()
[Test]
public async Task Parallelism_Limit_Larger_Than_Item_Count_Completes_Normally()
{
await using var rateLimited = Enumerable.Range(0, 5).ToList()
await using var bounded = Enumerable.Range(0, 5).ToList()
.ForEachAsync(_ => Task.CompletedTask)
.ProcessInParallel(100);
await rateLimited.WaitAsync();
await bounded.WaitAsync();

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

await Assert.That(rateLimited.GetEnumerableTasks().Count(x => x.IsCompletedSuccessfully)).IsEqualTo(5);
await Assert.That(bounded.GetEnumerableTasks().Count(x => x.IsCompletedSuccessfully)).IsEqualTo(5);
await Assert.That(results.Length).IsEqualTo(5);
}

Expand Down
46 changes: 6 additions & 40 deletions EnumerableAsyncProcessor/Builders/ActionAsyncProcessorBuilder.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,52 +29,18 @@ public IAsyncProcessor ProcessInBatches(int batchSize)
return new BatchAsyncProcessor(batchSize, _count, _taskSelector, _cancellationTokenSource).StartProcessing();
}

public IAsyncProcessor ProcessInParallel(int levelOfParallelism)
public IAsyncProcessor ProcessInParallel(int maxConcurrency, TimeSpan timeSpan)
{
return new RateLimitedParallelAsyncProcessor(_count, _taskSelector, levelOfParallelism, _cancellationTokenSource).StartProcessing();
}

public IAsyncProcessor ProcessInParallel(int levelOfParallelism, TimeSpan timeSpan)
{
return new TimedRateLimitedParallelAsyncProcessor(_count, _taskSelector, levelOfParallelism, timeSpan, _cancellationTokenSource).StartProcessing();
}

/// <summary>
/// Process tasks in parallel without concurrency limits.
/// </summary>
/// <returns>An async processor configured for parallel execution.</returns>
public IAsyncProcessor ProcessInParallel()
{
return ProcessInParallel(null, false);
}

/// <summary>
/// Process tasks in parallel without concurrency limits.
/// </summary>
/// <param name="scheduleOnThreadPool">If true, schedules tasks on thread pool to prevent blocking. Default is false for maximum performance.</param>
/// <returns>An async processor configured for parallel execution.</returns>
public IAsyncProcessor ProcessInParallel(bool scheduleOnThreadPool)
{
return ProcessInParallel(null, scheduleOnThreadPool);
}

/// <summary>
/// Process tasks in parallel with specified concurrency limit.
/// </summary>
/// <param name="maxConcurrency">Maximum concurrent operations.</param>
/// <returns>An async processor configured for parallel execution.</returns>
public IAsyncProcessor ProcessInParallel(int? maxConcurrency)
{
return ProcessInParallel(maxConcurrency, false);
return new TimedRateLimitedParallelAsyncProcessor(_count, _taskSelector, maxConcurrency, timeSpan, _cancellationTokenSource).StartProcessing();
}

/// <summary>
/// Process tasks in parallel with specified concurrency limit.
/// Process tasks in parallel, optionally limiting concurrency.
/// </summary>
/// <param name="maxConcurrency">Maximum concurrent operations.</param>
/// <param name="scheduleOnThreadPool">If true, schedules tasks on thread pool to prevent blocking.</param>
/// <param name="maxConcurrency">Maximum concurrent operations, or null for unbounded concurrency.</param>
/// <param name="scheduleOnThreadPool">For unbounded processing, schedules tasks on the thread pool when true.</param>
/// <returns>An async processor configured for parallel execution.</returns>
public IAsyncProcessor ProcessInParallel(int? maxConcurrency, bool scheduleOnThreadPool)
public IAsyncProcessor ProcessInParallel(int? maxConcurrency = null, bool scheduleOnThreadPool = false)
{
return new ParallelAsyncProcessor(_count, _taskSelector, _cancellationTokenSource, maxConcurrency, scheduleOnThreadPool).StartProcessing();
Comment thread
greptile-apps[bot] marked this conversation as resolved.
}
Expand Down
46 changes: 6 additions & 40 deletions EnumerableAsyncProcessor/Builders/ActionAsyncProcessorBuilder_1.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,52 +29,18 @@ public IAsyncProcessor<TOutput> ProcessInBatches(int batchSize)
return new ResultBatchAsyncProcessor<TOutput>(batchSize, _count, _taskSelector, _cancellationTokenSource).StartProcessing();
}

public IAsyncProcessor<TOutput> ProcessInParallel(int levelOfParallelism)
public IAsyncProcessor<TOutput> ProcessInParallel(int maxConcurrency, TimeSpan timeSpan)
{
return new ResultRateLimitedParallelAsyncProcessor<TOutput>(_count, _taskSelector, levelOfParallelism, _cancellationTokenSource).StartProcessing();
}

public IAsyncProcessor<TOutput> ProcessInParallel(int levelOfParallelism, TimeSpan timeSpan)
{
return new ResultTimedRateLimitedParallelAsyncProcessor<TOutput>(_count, _taskSelector, levelOfParallelism, timeSpan, _cancellationTokenSource).StartProcessing();
}

/// <summary>
/// Process tasks in parallel without concurrency limits and return results.
/// </summary>
/// <returns>An async processor configured for parallel execution that returns results.</returns>
public IAsyncProcessor<TOutput> ProcessInParallel()
{
return ProcessInParallel(null, false);
}

/// <summary>
/// Process tasks in parallel without concurrency limits and return results.
/// </summary>
/// <param name="scheduleOnThreadPool">If true, schedules tasks on thread pool to prevent blocking. Default is false for maximum performance.</param>
/// <returns>An async processor configured for parallel execution that returns results.</returns>
public IAsyncProcessor<TOutput> ProcessInParallel(bool scheduleOnThreadPool)
{
return ProcessInParallel(null, scheduleOnThreadPool);
}

/// <summary>
/// Process tasks in parallel with specified concurrency limit and return results.
/// </summary>
/// <param name="maxConcurrency">Maximum concurrent operations.</param>
/// <returns>An async processor configured for parallel execution that returns results.</returns>
public IAsyncProcessor<TOutput> ProcessInParallel(int? maxConcurrency)
{
return ProcessInParallel(maxConcurrency, false);
return new ResultTimedRateLimitedParallelAsyncProcessor<TOutput>(_count, _taskSelector, maxConcurrency, timeSpan, _cancellationTokenSource).StartProcessing();
}

/// <summary>
/// Process tasks in parallel with specified concurrency limit and return results.
/// Process tasks in parallel, optionally limiting concurrency, and return results.
/// </summary>
/// <param name="maxConcurrency">Maximum concurrent operations.</param>
/// <param name="scheduleOnThreadPool">If true, schedules tasks on thread pool to prevent blocking.</param>
/// <param name="maxConcurrency">Maximum concurrent operations, or null for unbounded concurrency.</param>
/// <param name="scheduleOnThreadPool">For unbounded processing, schedules tasks on the thread pool when true.</param>
/// <returns>An async processor configured for parallel execution that returns results.</returns>
public IAsyncProcessor<TOutput> ProcessInParallel(int? maxConcurrency, bool scheduleOnThreadPool)
public IAsyncProcessor<TOutput> ProcessInParallel(int? maxConcurrency = null, bool scheduleOnThreadPool = false)
{
return new ResultParallelAsyncProcessor<TOutput>(_count, _taskSelector, _cancellationTokenSource, maxConcurrency, scheduleOnThreadPool).StartProcessing();
}
Expand Down
47 changes: 4 additions & 43 deletions EnumerableAsyncProcessor/Builders/AsyncEnumerableActionAsyncProcessorBuilder_1.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,51 +30,12 @@ public AsyncEnumerableActionAsyncProcessorBuilder(
}

/// <summary>
/// Process items in parallel without concurrency limits.
/// Process items in parallel, optionally limiting concurrency.
/// </summary>
/// <param name="maxConcurrency">Maximum concurrent operations, or null for unbounded concurrency.</param>
/// <param name="scheduleOnThreadPool">For unbounded processing, schedules tasks on the thread pool when true.</param>
/// <returns>An async processor configured for parallel execution.</returns>
public IAsyncEnumerableProcessor ProcessInParallel()
{
return ProcessInParallel(null, false);
}

/// <summary>
/// Process items in parallel without concurrency limits.
/// </summary>
/// <param name="scheduleOnThreadPool">If true, schedules tasks on thread pool to prevent blocking. Default is false for maximum performance.</param>
/// <returns>An async processor configured for parallel execution.</returns>
public IAsyncEnumerableProcessor ProcessInParallel(bool scheduleOnThreadPool)
{
return ProcessInParallel(null, scheduleOnThreadPool);
}

/// <summary>
/// Process items in parallel with specified concurrency limit.
/// </summary>
/// <param name="maxConcurrency">Maximum concurrent operations.</param>
/// <returns>An async processor configured for parallel execution.</returns>
public IAsyncEnumerableProcessor ProcessInParallel(int maxConcurrency)
{
return ProcessInParallel((int?)maxConcurrency, false);
}

/// <summary>
/// Process items in parallel with specified concurrency limit.
/// </summary>
/// <param name="maxConcurrency">Maximum concurrent operations.</param>
/// <returns>An async processor configured for parallel execution.</returns>
public IAsyncEnumerableProcessor ProcessInParallel(int? maxConcurrency)
{
return ProcessInParallel(maxConcurrency, false);
}

/// <summary>
/// Process items in parallel with specified concurrency limit.
/// </summary>
/// <param name="maxConcurrency">Maximum concurrent operations.</param>
/// <param name="scheduleOnThreadPool">If true, schedules tasks on thread pool to prevent blocking.</param>
/// <returns>An async processor configured for parallel execution.</returns>
public IAsyncEnumerableProcessor ProcessInParallel(int? maxConcurrency, bool scheduleOnThreadPool)
public IAsyncEnumerableProcessor ProcessInParallel(int? maxConcurrency = null, bool scheduleOnThreadPool = false)
{
return new AsyncEnumerableParallelProcessor<TInput>(
_items, _taskSelector, maxConcurrency, scheduleOnThreadPool, _cancellationTokenSource);
Expand Down
47 changes: 4 additions & 43 deletions EnumerableAsyncProcessor/Builders/AsyncEnumerableActionAsyncProcessorBuilder_2.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,51 +30,12 @@ public AsyncEnumerableActionAsyncProcessorBuilder(
}

/// <summary>
/// Process items in parallel without concurrency limits and return results.
/// Process items in parallel, optionally limiting concurrency, and return results.
/// </summary>
/// <param name="maxConcurrency">Maximum concurrent operations, or null for unbounded concurrency.</param>
/// <param name="scheduleOnThreadPool">For unbounded processing, schedules tasks on the thread pool when true.</param>
/// <returns>An async processor configured for parallel execution.</returns>
public IAsyncEnumerableProcessor<TOutput> ProcessInParallel()
{
return ProcessInParallel(null, false);
}

/// <summary>
/// Process items in parallel without concurrency limits and return results.
/// </summary>
/// <param name="scheduleOnThreadPool">If true, schedules tasks on thread pool to prevent blocking. Default is false for maximum performance.</param>
/// <returns>An async processor configured for parallel execution.</returns>
public IAsyncEnumerableProcessor<TOutput> ProcessInParallel(bool scheduleOnThreadPool)
{
return ProcessInParallel(null, scheduleOnThreadPool);
}

/// <summary>
/// Process items in parallel with specified concurrency limit and return results.
/// </summary>
/// <param name="maxConcurrency">Maximum concurrent operations.</param>
/// <returns>An async processor configured for parallel execution.</returns>
public IAsyncEnumerableProcessor<TOutput> ProcessInParallel(int maxConcurrency)
{
return ProcessInParallel((int?)maxConcurrency, false);
}

/// <summary>
/// Process items in parallel with specified concurrency limit and return results.
/// </summary>
/// <param name="maxConcurrency">Maximum concurrent operations.</param>
/// <returns>An async processor configured for parallel execution.</returns>
public IAsyncEnumerableProcessor<TOutput> ProcessInParallel(int? maxConcurrency)
{
return ProcessInParallel(maxConcurrency, false);
}

/// <summary>
/// Process items in parallel with specified concurrency limit and return results.
/// </summary>
/// <param name="maxConcurrency">Maximum concurrent operations.</param>
/// <param name="scheduleOnThreadPool">If true, schedules tasks on thread pool to prevent blocking.</param>
/// <returns>An async processor configured for parallel execution.</returns>
public IAsyncEnumerableProcessor<TOutput> ProcessInParallel(int? maxConcurrency, bool scheduleOnThreadPool)
public IAsyncEnumerableProcessor<TOutput> ProcessInParallel(int? maxConcurrency = null, bool scheduleOnThreadPool = false)
{
return new ResultAsyncEnumerableParallelProcessor<TInput, TOutput>(
_items, _taskSelector, maxConcurrency, scheduleOnThreadPool, _cancellationTokenSource);
Expand Down
47 changes: 6 additions & 41 deletions EnumerableAsyncProcessor/Builders/ItemActionAsyncProcessorBuilder_1.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,54 +30,19 @@ public IAsyncProcessor ProcessInBatches(int batchSize)
.StartProcessing();
}

public IAsyncProcessor ProcessInParallel(int levelOfParallelism)
public IAsyncProcessor ProcessInParallel(int maxConcurrency, TimeSpan timeSpan)
{
return new RateLimitedParallelAsyncProcessor<TInput>(_items, _taskSelector, levelOfParallelism, _cancellationTokenSource)
return new TimedRateLimitedParallelAsyncProcessor<TInput>(_items, _taskSelector, maxConcurrency, timeSpan, _cancellationTokenSource)
.StartProcessing();
}

public IAsyncProcessor ProcessInParallel(int levelOfParallelism, TimeSpan timeSpan)
{
return new TimedRateLimitedParallelAsyncProcessor<TInput>(_items, _taskSelector, levelOfParallelism, timeSpan, _cancellationTokenSource)
.StartProcessing();
}

/// <summary>
/// Process items in parallel without concurrency limits.
/// </summary>
/// <returns>An async processor configured for parallel execution.</returns>
public IAsyncProcessor ProcessInParallel()
{
return ProcessInParallel(null, false);
}

/// <summary>
/// Process items in parallel without concurrency limits.
/// </summary>
/// <param name="scheduleOnThreadPool">If true, schedules tasks on thread pool to prevent blocking. Default is false for maximum performance.</param>
/// <returns>An async processor configured for parallel execution.</returns>
public IAsyncProcessor ProcessInParallel(bool scheduleOnThreadPool)
{
return ProcessInParallel(null, scheduleOnThreadPool);
}

/// <summary>
/// Process items in parallel with specified concurrency limit.
/// </summary>
/// <param name="maxConcurrency">Maximum concurrent operations.</param>
/// <returns>An async processor configured for parallel execution.</returns>
public IAsyncProcessor ProcessInParallel(int? maxConcurrency)
{
return ProcessInParallel(maxConcurrency, false);
}

/// <summary>
/// Process items in parallel with specified concurrency limit.
/// Process items in parallel, optionally limiting concurrency.
/// </summary>
/// <param name="maxConcurrency">Maximum concurrent operations.</param>
/// <param name="scheduleOnThreadPool">If true, schedules tasks on thread pool to prevent blocking.</param>
/// <param name="maxConcurrency">Maximum concurrent operations, or null for unbounded concurrency.</param>
/// <param name="scheduleOnThreadPool">For unbounded processing, schedules tasks on the thread pool when true.</param>
/// <returns>An async processor configured for parallel execution.</returns>
public IAsyncProcessor ProcessInParallel(int? maxConcurrency, bool scheduleOnThreadPool)
public IAsyncProcessor ProcessInParallel(int? maxConcurrency = null, bool scheduleOnThreadPool = false)
{
return new ParallelAsyncProcessor<TInput>(_items, _taskSelector, _cancellationTokenSource, maxConcurrency, scheduleOnThreadPool)
.StartProcessing();
Expand Down
Loading
Loading