diff --git a/EnumerableAsyncProcessor.UnitTests/RateLimitedParallelAsyncProcessorTests.cs b/EnumerableAsyncProcessor.UnitTests/BoundedParallelAsyncProcessorTests.cs similarity index 99% rename from EnumerableAsyncProcessor.UnitTests/RateLimitedParallelAsyncProcessorTests.cs rename to EnumerableAsyncProcessor.UnitTests/BoundedParallelAsyncProcessorTests.cs index b6ccdaa..143e873 100644 --- a/EnumerableAsyncProcessor.UnitTests/RateLimitedParallelAsyncProcessorTests.cs +++ b/EnumerableAsyncProcessor.UnitTests/BoundedParallelAsyncProcessorTests.cs @@ -8,7 +8,7 @@ namespace EnumerableAsyncProcessor.UnitTests; -public class RateLimitedParallelAsyncProcessorTests +public class BoundedParallelAsyncProcessorTests { [MatrixDataSource] [Test, Repeat(5)] @@ -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); } -} \ No newline at end of file +} diff --git a/EnumerableAsyncProcessor.UnitTests/ValidationRegressionTests.cs b/EnumerableAsyncProcessor.UnitTests/ValidationRegressionTests.cs index 8e70e93..9fbe11a 100644 --- a/EnumerableAsyncProcessor.UnitTests/ValidationRegressionTests.cs +++ b/EnumerableAsyncProcessor.UnitTests/ValidationRegressionTests.cs @@ -31,7 +31,7 @@ await AssertThrows(() => } [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(() => new[] { 1 }.ForEachAsync(_ => Task.CompletedTask).ProcessInParallel(0)); diff --git a/EnumerableAsyncProcessor.UnitTests/WorkerPoolBehaviourTests.cs b/EnumerableAsyncProcessor.UnitTests/WorkerPoolBehaviourTests.cs index 0148a4f..5aa469d 100644 --- a/EnumerableAsyncProcessor.UnitTests/WorkerPoolBehaviourTests.cs +++ b/EnumerableAsyncProcessor.UnitTests/WorkerPoolBehaviourTests.cs @@ -4,17 +4,32 @@ using System.Threading; using System.Threading.Tasks; using EnumerableAsyncProcessor.Extensions; +using EnumerableAsyncProcessor.RunnableProcessors; namespace EnumerableAsyncProcessor.UnitTests; /// -/// 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. /// 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)); + await Assert.That(named.GetType()).IsEqualTo(typeof(ParallelAsyncProcessor)); + } + [Test, Repeat(3)] public async Task MaxConcurrency_Path_Obeys_The_Limit_And_Processes_Everything(CancellationToken cancellationToken) { @@ -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; @@ -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); } diff --git a/EnumerableAsyncProcessor/Builders/ActionAsyncProcessorBuilder.cs b/EnumerableAsyncProcessor/Builders/ActionAsyncProcessorBuilder.cs index 5cae366..9c32159 100644 --- a/EnumerableAsyncProcessor/Builders/ActionAsyncProcessorBuilder.cs +++ b/EnumerableAsyncProcessor/Builders/ActionAsyncProcessorBuilder.cs @@ -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(); - } - - /// - /// Process tasks in parallel without concurrency limits. - /// - /// An async processor configured for parallel execution. - public IAsyncProcessor ProcessInParallel() - { - return ProcessInParallel(null, false); - } - - /// - /// Process tasks in parallel without concurrency limits. - /// - /// If true, schedules tasks on thread pool to prevent blocking. Default is false for maximum performance. - /// An async processor configured for parallel execution. - public IAsyncProcessor ProcessInParallel(bool scheduleOnThreadPool) - { - return ProcessInParallel(null, scheduleOnThreadPool); - } - - /// - /// Process tasks in parallel with specified concurrency limit. - /// - /// Maximum concurrent operations. - /// An async processor configured for parallel execution. - public IAsyncProcessor ProcessInParallel(int? maxConcurrency) - { - return ProcessInParallel(maxConcurrency, false); + return new TimedRateLimitedParallelAsyncProcessor(_count, _taskSelector, maxConcurrency, timeSpan, _cancellationTokenSource).StartProcessing(); } /// - /// Process tasks in parallel with specified concurrency limit. + /// Process tasks in parallel, optionally limiting concurrency. /// - /// Maximum concurrent operations. - /// If true, schedules tasks on thread pool to prevent blocking. + /// Maximum concurrent operations, or null for unbounded concurrency. + /// For unbounded processing, schedules tasks on the thread pool when true. /// An async processor configured for parallel execution. - 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(); } diff --git a/EnumerableAsyncProcessor/Builders/ActionAsyncProcessorBuilder_1.cs b/EnumerableAsyncProcessor/Builders/ActionAsyncProcessorBuilder_1.cs index 5a6d32c..9eb3a91 100644 --- a/EnumerableAsyncProcessor/Builders/ActionAsyncProcessorBuilder_1.cs +++ b/EnumerableAsyncProcessor/Builders/ActionAsyncProcessorBuilder_1.cs @@ -29,52 +29,18 @@ public IAsyncProcessor ProcessInBatches(int batchSize) return new ResultBatchAsyncProcessor(batchSize, _count, _taskSelector, _cancellationTokenSource).StartProcessing(); } - public IAsyncProcessor ProcessInParallel(int levelOfParallelism) + public IAsyncProcessor ProcessInParallel(int maxConcurrency, TimeSpan timeSpan) { - return new ResultRateLimitedParallelAsyncProcessor(_count, _taskSelector, levelOfParallelism, _cancellationTokenSource).StartProcessing(); - } - - public IAsyncProcessor ProcessInParallel(int levelOfParallelism, TimeSpan timeSpan) - { - return new ResultTimedRateLimitedParallelAsyncProcessor(_count, _taskSelector, levelOfParallelism, timeSpan, _cancellationTokenSource).StartProcessing(); - } - - /// - /// Process tasks in parallel without concurrency limits and return results. - /// - /// An async processor configured for parallel execution that returns results. - public IAsyncProcessor ProcessInParallel() - { - return ProcessInParallel(null, false); - } - - /// - /// Process tasks in parallel without concurrency limits and return results. - /// - /// If true, schedules tasks on thread pool to prevent blocking. Default is false for maximum performance. - /// An async processor configured for parallel execution that returns results. - public IAsyncProcessor ProcessInParallel(bool scheduleOnThreadPool) - { - return ProcessInParallel(null, scheduleOnThreadPool); - } - - /// - /// Process tasks in parallel with specified concurrency limit and return results. - /// - /// Maximum concurrent operations. - /// An async processor configured for parallel execution that returns results. - public IAsyncProcessor ProcessInParallel(int? maxConcurrency) - { - return ProcessInParallel(maxConcurrency, false); + return new ResultTimedRateLimitedParallelAsyncProcessor(_count, _taskSelector, maxConcurrency, timeSpan, _cancellationTokenSource).StartProcessing(); } /// - /// Process tasks in parallel with specified concurrency limit and return results. + /// Process tasks in parallel, optionally limiting concurrency, and return results. /// - /// Maximum concurrent operations. - /// If true, schedules tasks on thread pool to prevent blocking. + /// Maximum concurrent operations, or null for unbounded concurrency. + /// For unbounded processing, schedules tasks on the thread pool when true. /// An async processor configured for parallel execution that returns results. - public IAsyncProcessor ProcessInParallel(int? maxConcurrency, bool scheduleOnThreadPool) + public IAsyncProcessor ProcessInParallel(int? maxConcurrency = null, bool scheduleOnThreadPool = false) { return new ResultParallelAsyncProcessor(_count, _taskSelector, _cancellationTokenSource, maxConcurrency, scheduleOnThreadPool).StartProcessing(); } diff --git a/EnumerableAsyncProcessor/Builders/AsyncEnumerableActionAsyncProcessorBuilder_1.cs b/EnumerableAsyncProcessor/Builders/AsyncEnumerableActionAsyncProcessorBuilder_1.cs index 679eb69..a993ac3 100644 --- a/EnumerableAsyncProcessor/Builders/AsyncEnumerableActionAsyncProcessorBuilder_1.cs +++ b/EnumerableAsyncProcessor/Builders/AsyncEnumerableActionAsyncProcessorBuilder_1.cs @@ -30,51 +30,12 @@ public AsyncEnumerableActionAsyncProcessorBuilder( } /// - /// Process items in parallel without concurrency limits. + /// Process items in parallel, optionally limiting concurrency. /// + /// Maximum concurrent operations, or null for unbounded concurrency. + /// For unbounded processing, schedules tasks on the thread pool when true. /// An async processor configured for parallel execution. - public IAsyncEnumerableProcessor ProcessInParallel() - { - return ProcessInParallel(null, false); - } - - /// - /// Process items in parallel without concurrency limits. - /// - /// If true, schedules tasks on thread pool to prevent blocking. Default is false for maximum performance. - /// An async processor configured for parallel execution. - public IAsyncEnumerableProcessor ProcessInParallel(bool scheduleOnThreadPool) - { - return ProcessInParallel(null, scheduleOnThreadPool); - } - - /// - /// Process items in parallel with specified concurrency limit. - /// - /// Maximum concurrent operations. - /// An async processor configured for parallel execution. - public IAsyncEnumerableProcessor ProcessInParallel(int maxConcurrency) - { - return ProcessInParallel((int?)maxConcurrency, false); - } - - /// - /// Process items in parallel with specified concurrency limit. - /// - /// Maximum concurrent operations. - /// An async processor configured for parallel execution. - public IAsyncEnumerableProcessor ProcessInParallel(int? maxConcurrency) - { - return ProcessInParallel(maxConcurrency, false); - } - - /// - /// Process items in parallel with specified concurrency limit. - /// - /// Maximum concurrent operations. - /// If true, schedules tasks on thread pool to prevent blocking. - /// An async processor configured for parallel execution. - public IAsyncEnumerableProcessor ProcessInParallel(int? maxConcurrency, bool scheduleOnThreadPool) + public IAsyncEnumerableProcessor ProcessInParallel(int? maxConcurrency = null, bool scheduleOnThreadPool = false) { return new AsyncEnumerableParallelProcessor( _items, _taskSelector, maxConcurrency, scheduleOnThreadPool, _cancellationTokenSource); diff --git a/EnumerableAsyncProcessor/Builders/AsyncEnumerableActionAsyncProcessorBuilder_2.cs b/EnumerableAsyncProcessor/Builders/AsyncEnumerableActionAsyncProcessorBuilder_2.cs index 2a3f236..edfd1b1 100644 --- a/EnumerableAsyncProcessor/Builders/AsyncEnumerableActionAsyncProcessorBuilder_2.cs +++ b/EnumerableAsyncProcessor/Builders/AsyncEnumerableActionAsyncProcessorBuilder_2.cs @@ -30,51 +30,12 @@ public AsyncEnumerableActionAsyncProcessorBuilder( } /// - /// Process items in parallel without concurrency limits and return results. + /// Process items in parallel, optionally limiting concurrency, and return results. /// + /// Maximum concurrent operations, or null for unbounded concurrency. + /// For unbounded processing, schedules tasks on the thread pool when true. /// An async processor configured for parallel execution. - public IAsyncEnumerableProcessor ProcessInParallel() - { - return ProcessInParallel(null, false); - } - - /// - /// Process items in parallel without concurrency limits and return results. - /// - /// If true, schedules tasks on thread pool to prevent blocking. Default is false for maximum performance. - /// An async processor configured for parallel execution. - public IAsyncEnumerableProcessor ProcessInParallel(bool scheduleOnThreadPool) - { - return ProcessInParallel(null, scheduleOnThreadPool); - } - - /// - /// Process items in parallel with specified concurrency limit and return results. - /// - /// Maximum concurrent operations. - /// An async processor configured for parallel execution. - public IAsyncEnumerableProcessor ProcessInParallel(int maxConcurrency) - { - return ProcessInParallel((int?)maxConcurrency, false); - } - - /// - /// Process items in parallel with specified concurrency limit and return results. - /// - /// Maximum concurrent operations. - /// An async processor configured for parallel execution. - public IAsyncEnumerableProcessor ProcessInParallel(int? maxConcurrency) - { - return ProcessInParallel(maxConcurrency, false); - } - - /// - /// Process items in parallel with specified concurrency limit and return results. - /// - /// Maximum concurrent operations. - /// If true, schedules tasks on thread pool to prevent blocking. - /// An async processor configured for parallel execution. - public IAsyncEnumerableProcessor ProcessInParallel(int? maxConcurrency, bool scheduleOnThreadPool) + public IAsyncEnumerableProcessor ProcessInParallel(int? maxConcurrency = null, bool scheduleOnThreadPool = false) { return new ResultAsyncEnumerableParallelProcessor( _items, _taskSelector, maxConcurrency, scheduleOnThreadPool, _cancellationTokenSource); diff --git a/EnumerableAsyncProcessor/Builders/ItemActionAsyncProcessorBuilder_1.cs b/EnumerableAsyncProcessor/Builders/ItemActionAsyncProcessorBuilder_1.cs index 356e584..3ac2223 100644 --- a/EnumerableAsyncProcessor/Builders/ItemActionAsyncProcessorBuilder_1.cs +++ b/EnumerableAsyncProcessor/Builders/ItemActionAsyncProcessorBuilder_1.cs @@ -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(_items, _taskSelector, levelOfParallelism, _cancellationTokenSource) + return new TimedRateLimitedParallelAsyncProcessor(_items, _taskSelector, maxConcurrency, timeSpan, _cancellationTokenSource) .StartProcessing(); } - public IAsyncProcessor ProcessInParallel(int levelOfParallelism, TimeSpan timeSpan) - { - return new TimedRateLimitedParallelAsyncProcessor(_items, _taskSelector, levelOfParallelism, timeSpan, _cancellationTokenSource) - .StartProcessing(); - } - - /// - /// Process items in parallel without concurrency limits. - /// - /// An async processor configured for parallel execution. - public IAsyncProcessor ProcessInParallel() - { - return ProcessInParallel(null, false); - } - - /// - /// Process items in parallel without concurrency limits. - /// - /// If true, schedules tasks on thread pool to prevent blocking. Default is false for maximum performance. - /// An async processor configured for parallel execution. - public IAsyncProcessor ProcessInParallel(bool scheduleOnThreadPool) - { - return ProcessInParallel(null, scheduleOnThreadPool); - } - - /// - /// Process items in parallel with specified concurrency limit. - /// - /// Maximum concurrent operations. - /// An async processor configured for parallel execution. - public IAsyncProcessor ProcessInParallel(int? maxConcurrency) - { - return ProcessInParallel(maxConcurrency, false); - } - /// - /// Process items in parallel with specified concurrency limit. + /// Process items in parallel, optionally limiting concurrency. /// - /// Maximum concurrent operations. - /// If true, schedules tasks on thread pool to prevent blocking. + /// Maximum concurrent operations, or null for unbounded concurrency. + /// For unbounded processing, schedules tasks on the thread pool when true. /// An async processor configured for parallel execution. - public IAsyncProcessor ProcessInParallel(int? maxConcurrency, bool scheduleOnThreadPool) + public IAsyncProcessor ProcessInParallel(int? maxConcurrency = null, bool scheduleOnThreadPool = false) { return new ParallelAsyncProcessor(_items, _taskSelector, _cancellationTokenSource, maxConcurrency, scheduleOnThreadPool) .StartProcessing(); diff --git a/EnumerableAsyncProcessor/Builders/ItemActionAsyncProcessorBuilder_2.cs b/EnumerableAsyncProcessor/Builders/ItemActionAsyncProcessorBuilder_2.cs index 886098d..fbad93f 100644 --- a/EnumerableAsyncProcessor/Builders/ItemActionAsyncProcessorBuilder_2.cs +++ b/EnumerableAsyncProcessor/Builders/ItemActionAsyncProcessorBuilder_2.cs @@ -40,24 +40,9 @@ public IAsyncProcessor ProcessInBatches(int batchSize) } /// - /// Processes items in parallel with the specified level of parallelism. + /// Processes items in parallel with the specified concurrency and time constraints. /// - /// The maximum number of concurrent operations. - /// An async processor that implements IDisposable and IAsyncDisposable. - /// Use 'await using' or proper disposal to ensure resources are cleaned up. - /// - /// The returned processor should be disposed to ensure proper cleanup of internal resources - /// and cancellation of running tasks. Use 'await using var processor = ...' for automatic disposal. - /// - public IAsyncProcessor ProcessInParallel(int levelOfParallelism) - { - return new ResultRateLimitedParallelAsyncProcessor(_items, _taskSelector, levelOfParallelism, _cancellationTokenSource).StartProcessing(); - } - - /// - /// Processes items in parallel with the specified level of parallelism and time constraints. - /// - /// The maximum number of concurrent operations. + /// The maximum number of concurrent operations. /// The time span constraint for rate limiting. /// An async processor that implements IDisposable and IAsyncDisposable. /// Use 'await using' or proper disposal to ensure resources are cleaned up. @@ -65,67 +50,23 @@ public IAsyncProcessor ProcessInParallel(int levelOfParallelism) /// The returned processor should be disposed to ensure proper cleanup of internal resources /// and cancellation of running tasks. Use 'await using var processor = ...' for automatic disposal. /// - public IAsyncProcessor ProcessInParallel(int levelOfParallelism, TimeSpan timeSpan) - { - return new ResultTimedRateLimitedParallelAsyncProcessor(_items, _taskSelector, levelOfParallelism, timeSpan, _cancellationTokenSource).StartProcessing(); - } - - /// - /// Process items in parallel without concurrency limits and return results. - /// - /// An async processor that implements IDisposable and IAsyncDisposable. - /// Use 'await using' or proper disposal to ensure resources are cleaned up. - /// - /// The returned processor should be disposed to ensure proper cleanup of internal resources - /// and cancellation of running tasks. Use 'await using var processor = ...' for automatic disposal. - /// - public IAsyncProcessor ProcessInParallel() - { - return ProcessInParallel(null, false); - } - - /// - /// Process items in parallel without concurrency limits and return results. - /// - /// If true, schedules tasks on thread pool to prevent blocking. Default is false for maximum performance. - /// An async processor that implements IDisposable and IAsyncDisposable. - /// Use 'await using' or proper disposal to ensure resources are cleaned up. - /// - /// The returned processor should be disposed to ensure proper cleanup of internal resources - /// and cancellation of running tasks. Use 'await using var processor = ...' for automatic disposal. - /// - public IAsyncProcessor ProcessInParallel(bool scheduleOnThreadPool) - { - return ProcessInParallel(null, scheduleOnThreadPool); - } - - /// - /// Process items in parallel with specified concurrency limit and return results. - /// - /// Maximum concurrent operations. - /// An async processor that implements IDisposable and IAsyncDisposable. - /// Use 'await using' or proper disposal to ensure resources are cleaned up. - /// - /// The returned processor should be disposed to ensure proper cleanup of internal resources - /// and cancellation of running tasks. Use 'await using var processor = ...' for automatic disposal. - /// - public IAsyncProcessor ProcessInParallel(int? maxConcurrency) + public IAsyncProcessor ProcessInParallel(int maxConcurrency, TimeSpan timeSpan) { - return ProcessInParallel(maxConcurrency, false); + return new ResultTimedRateLimitedParallelAsyncProcessor(_items, _taskSelector, maxConcurrency, timeSpan, _cancellationTokenSource).StartProcessing(); } /// - /// Process items in parallel with specified concurrency limit and return results. + /// Process items in parallel, optionally limiting concurrency, and return results. /// - /// Maximum concurrent operations. - /// If true, schedules tasks on thread pool to prevent blocking. + /// Maximum concurrent operations, or null for unbounded concurrency. + /// For unbounded processing, schedules tasks on the thread pool when true. /// An async processor that implements IDisposable and IAsyncDisposable. /// Use 'await using' or proper disposal to ensure resources are cleaned up. /// /// The returned processor should be disposed to ensure proper cleanup of internal resources /// and cancellation of running tasks. Use 'await using var processor = ...' for automatic disposal. /// - public IAsyncProcessor ProcessInParallel(int? maxConcurrency, bool scheduleOnThreadPool) + public IAsyncProcessor ProcessInParallel(int? maxConcurrency = null, bool scheduleOnThreadPool = false) { return new ResultParallelAsyncProcessor(_items, _taskSelector, _cancellationTokenSource, maxConcurrency, scheduleOnThreadPool).StartProcessing(); } diff --git a/EnumerableAsyncProcessor/RunnableProcessors/RateLimitedParallelAsyncProcessor.cs b/EnumerableAsyncProcessor/RunnableProcessors/RateLimitedParallelAsyncProcessor.cs deleted file mode 100644 index df8147e..0000000 --- a/EnumerableAsyncProcessor/RunnableProcessors/RateLimitedParallelAsyncProcessor.cs +++ /dev/null @@ -1,21 +0,0 @@ -using EnumerableAsyncProcessor.RunnableProcessors.Abstract; -using EnumerableAsyncProcessor.Validation; - -namespace EnumerableAsyncProcessor.RunnableProcessors; - -public class RateLimitedParallelAsyncProcessor : AbstractAsyncProcessor -{ - private readonly int _levelsOfParallelism; - - internal RateLimitedParallelAsyncProcessor(int count, Func taskSelector, int levelsOfParallelism, CancellationTokenSource cancellationTokenSource) : base(count, taskSelector, cancellationTokenSource) - { - ValidationHelper.ThrowIfNegativeOrZero(levelsOfParallelism); - - _levelsOfParallelism = levelsOfParallelism; - } - - internal override Task Process() - { - return WorkerPool.ProcessAsync(TaskWrappers, _levelsOfParallelism, minimumIterationTime: null, CancellationToken); - } -} diff --git a/EnumerableAsyncProcessor/RunnableProcessors/RateLimitedParallelAsyncProcessor_1.cs b/EnumerableAsyncProcessor/RunnableProcessors/RateLimitedParallelAsyncProcessor_1.cs deleted file mode 100644 index 32dab70..0000000 --- a/EnumerableAsyncProcessor/RunnableProcessors/RateLimitedParallelAsyncProcessor_1.cs +++ /dev/null @@ -1,21 +0,0 @@ -using EnumerableAsyncProcessor.RunnableProcessors.Abstract; -using EnumerableAsyncProcessor.Validation; - -namespace EnumerableAsyncProcessor.RunnableProcessors; - -public class RateLimitedParallelAsyncProcessor : AbstractAsyncProcessor -{ - private readonly int _levelsOfParallelism; - - internal RateLimitedParallelAsyncProcessor(IEnumerable items, Func taskSelector, int levelsOfParallelism, CancellationTokenSource cancellationTokenSource) : base(items, taskSelector, cancellationTokenSource) - { - ValidationHelper.ThrowIfNegativeOrZero(levelsOfParallelism); - - _levelsOfParallelism = levelsOfParallelism; - } - - internal override Task Process() - { - return WorkerPool.ProcessAsync(TaskWrappers, _levelsOfParallelism, minimumIterationTime: null, CancellationToken); - } -} diff --git a/EnumerableAsyncProcessor/RunnableProcessors/ResultProcessors/ResultRateLimitedParallelAsyncProcessor_1.cs b/EnumerableAsyncProcessor/RunnableProcessors/ResultProcessors/ResultRateLimitedParallelAsyncProcessor_1.cs deleted file mode 100644 index c42f06e..0000000 --- a/EnumerableAsyncProcessor/RunnableProcessors/ResultProcessors/ResultRateLimitedParallelAsyncProcessor_1.cs +++ /dev/null @@ -1,22 +0,0 @@ - -using EnumerableAsyncProcessor.RunnableProcessors.ResultProcessors.Abstract; -using EnumerableAsyncProcessor.Validation; - -namespace EnumerableAsyncProcessor.RunnableProcessors.ResultProcessors; - -public class ResultRateLimitedParallelAsyncProcessor : ResultAbstractAsyncProcessor -{ - private readonly int _levelsOfParallelism; - - internal ResultRateLimitedParallelAsyncProcessor(int count, Func> taskSelector, int levelsOfParallelism, CancellationTokenSource cancellationTokenSource) : base(count, taskSelector, cancellationTokenSource) - { - ValidationHelper.ThrowIfNegativeOrZero(levelsOfParallelism); - - _levelsOfParallelism = levelsOfParallelism; - } - - internal override Task Process() - { - return WorkerPool.ProcessAsync(TaskWrappers, _levelsOfParallelism, minimumIterationTime: null, CancellationToken); - } -} diff --git a/EnumerableAsyncProcessor/RunnableProcessors/ResultProcessors/ResultRateLimitedParallelAsyncProcessor_2.cs b/EnumerableAsyncProcessor/RunnableProcessors/ResultProcessors/ResultRateLimitedParallelAsyncProcessor_2.cs deleted file mode 100644 index 5753020..0000000 --- a/EnumerableAsyncProcessor/RunnableProcessors/ResultProcessors/ResultRateLimitedParallelAsyncProcessor_2.cs +++ /dev/null @@ -1,21 +0,0 @@ -using EnumerableAsyncProcessor.RunnableProcessors.ResultProcessors.Abstract; -using EnumerableAsyncProcessor.Validation; - -namespace EnumerableAsyncProcessor.RunnableProcessors.ResultProcessors; - -public class ResultRateLimitedParallelAsyncProcessor : ResultAbstractAsyncProcessor -{ - private readonly int _levelsOfParallelism; - - internal ResultRateLimitedParallelAsyncProcessor(IEnumerable items, Func> taskSelector, int levelsOfParallelism, CancellationTokenSource cancellationTokenSource) : base(items, taskSelector, cancellationTokenSource) - { - ValidationHelper.ThrowIfNegativeOrZero(levelsOfParallelism); - - _levelsOfParallelism = levelsOfParallelism; - } - - internal override Task Process() - { - return WorkerPool.ProcessAsync(TaskWrappers, _levelsOfParallelism, minimumIterationTime: null, CancellationToken); - } -} diff --git a/README.md b/README.md index 179f1bd..2419630 100644 --- a/README.md +++ b/README.md @@ -28,23 +28,19 @@ Maybe you want it slow. Maybe you want it at a safe balance. Maybe you just don't want to write all the boilerplate code that comes with managing asynchronous operations! -### Rate Limited Parallel Processor +### Parallel Processor (Optional Concurrency Limit) **Types** -| Type | Source Object | Return Object | Method 1 | Method 2 | -|--------------------------------------------------|---------------|---------------|--------------------| ------------------ | -| `RateLimitedParallelAsyncProcessor` | ❌ | ❌ | `.WithExecutionCount(int)` | `.ForEachAsync(delegate)` | -| `RateLimitedParallelAsyncProcessor` | ✔ | ❌ | `.WithItems(IEnumerable)` | `.ForEachAsync(delegate)` | -| `ResultRateLimitedParallelAsyncProcessor` | ❌ | ✔ | `.WithExecutionCount(int)` | `.SelectAsync(delegate)` | -| `ResultRateLimitedParallelAsyncProcessor` | ✔ | ✔ | `.WithItems(IEnumerable)` | `.SelectAsync(delegate)` | +| Type | Source Object | Return Object | Method 1 | Method 2 | +|--------------------------------------------------|---------------|---------------|---------------------|--------------------| +| `ParallelAsyncProcessor` | ❌ | ❌ | `.WithExecutionCount(int)` | `.ForEachAsync(delegate)` | +| `ParallelAsyncProcessor` | ✔ | ❌ | `.WithItems(IEnumerable)` | `.ForEachAsync(delegate)` | +| `ResultParallelAsyncProcessor` | ❌ | ✔ | `.WithExecutionCount(int)` | `.SelectAsync(delegate)` | +| `ResultParallelAsyncProcessor` | ✔ | ✔ | `.WithItems(IEnumerable)` | `.SelectAsync(delegate)` | **How it works** -Processes your Asynchronous Tasks in Parallel, but honouring the limit that you set. As one finishes, another will start. - -E.g. If you set a limit of 100, only 100 should ever run at any one time - -This is a hybrid between Parallel Processor and Batch Processor (see below) - Trying to address the caveats of both. Increasing the speed of batching, but not overwhelming the system by using full parallelisation. +Processes asynchronous tasks in parallel. Pass `maxConcurrency` to bound the number of operations running at once, or omit it for unbounded concurrency. **Usage** @@ -54,16 +50,24 @@ var ids = Enumerable.Range(0, 5000).ToList(); // SelectAsync for if you want to return something - using proper disposal await using var processor = ids .SelectAsync(id => DoSomethingAndReturnSomethingAsync(id), CancellationToken.None) - .ProcessInParallel(levelOfParallelism: 100); + .ProcessInParallel(maxConcurrency: 100); var results = await processor.GetResultsAsync(); // ForEachAsync for when you have nothing to return - using proper disposal await using var voidProcessor = ids .ForEachAsync(id => DoSomethingAsync(id), CancellationToken.None) - .ProcessInParallel(levelOfParallelism: 100); + .ProcessInParallel(maxConcurrency: 100); await voidProcessor.WaitAsync(); + +// Omit maxConcurrency for unbounded parallel processing +await using var unboundedProcessor = ids + .ForEachAsync(id => DoSomethingAsync(id), CancellationToken.None) + .ProcessInParallel(); +await unboundedProcessor.WaitAsync(); ``` +Choose a concurrency limit that protects downstream resources. Unbounded processing can increase memory, CPU, and network pressure. + ### Timed Rate Limited Parallel Processor (e.g. Limit RPS) **Types** @@ -90,13 +94,13 @@ var ids = Enumerable.Range(0, 5000).ToList(); // SelectAsync for if you want to return something - using proper disposal await using var processor = ids .SelectAsync(id => DoSomethingAndReturnSomethingAsync(id), CancellationToken.None) - .ProcessInParallel(levelOfParallelism: 100, TimeSpan.FromSeconds(1)); + .ProcessInParallel(maxConcurrency: 100, TimeSpan.FromSeconds(1)); var results = await processor.GetResultsAsync(); // ForEachAsync for when you have nothing to return - using proper disposal await using var voidProcessor = ids .ForEachAsync(id => DoSomethingAsync(id), CancellationToken.None) - .ProcessInParallel(levelOfParallelism: 100, TimeSpan.FromSeconds(1)); + .ProcessInParallel(maxConcurrency: 100, TimeSpan.FromSeconds(1)); await voidProcessor.WaitAsync(); ``` @@ -173,40 +177,6 @@ await ids - If even just 1 Task in a batch is slow or hangs, this will prevent the next batch from starting - If you set a batch of 100, and 70 have finished, you'll only have 30 left executing. This could slow things down -### Parallel - -**Types** - -| Type | Source Object | Return Object | Method 1 | Method 2 | -|--------------------------------------------------|---------------|---------------|--------------------| ------------------ | -| `ParallelAsyncProcessor` | ❌ | ❌ | `.WithExecutionCount(int)` | `.ForEachAsync(delegate)` | -| `ParallelAsyncProcessor` | ✔ | ❌ | `.WithItems(IEnumerable)` | `.ForEachAsync(delegate)` | -| `ResultParallelAsyncProcessor` | ❌ | ✔ | `.WithExecutionCount(int)` | `.SelectAsync(delegate)` | -| `ResultParallelAsyncProcessor` | ✔ | ✔ | `.WithItems(IEnumerable)` | `.SelectAsync(delegate)` | - -**How it works** -Processes your Asynchronous Tasks as fast as it can. All at the same time if it can - -**Usage** - -```csharp -var ids = Enumerable.Range(0, 5000).ToList(); - -// SelectAsync for if you want to return something -var results = await ids - .SelectAsync(id => DoSomethingAndReturnSomethingAsync(id), CancellationToken.None) - .ProcessInParallel(); - -// ForEachAsync for when you have nothing to return -await ids - .ForEachAsync(id => DoSomethingAsync(id), CancellationToken.None) - .ProcessInParallel(); -``` - -**Caveats** - -- Depending on how many operations you have, you could overwhelm your system. Memory and CPU and Network usage could spike, and cause bottlenecks / crashes / exceptions - ### Processor Methods As above, you can see that you can just `await` on the processor to get the results.