EnumerableAsyncProcessor

Process Multiple Asynchronous Tasks in Various Ways - One at a time / Batched / Rate limited / Concurrently

<!--  -->

Installation

Install via Nuget Install-Package EnumerableAsyncProcessor

Why I built this

Because I've come across situations where you need to fine tune the rate at which you do things. Maybe you want it fast. 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

Types

| Type | Source Object | Return Object | Method 1 | Method 2 | |--------------------------------------------------|---------------|---------------|--------------------| ------------------ | | RateLimitedParallelAsyncProcessor | ❌ | ❌ | .WithExecutionCount(int) | .ForEachAsync(delegate) | | RateLimitedParallelAsyncProcessor<TInput> | ✔ | ❌ | .WithItems(IEnumerable<TInput>) | .ForEachAsync(delegate) | | ResultRateLimitedParallelAsyncProcessor<TOutput> | ❌ | ✔ | .WithExecutionCount(int) | .SelectAsync(delegate) | | ResultRateLimitedParallelAsyncProcessor<TInput, TOutput> | ✔ | ✔ | .WithItems(IEnumerable<TInput>) | .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.

Usage

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);
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);
await voidProcessor.WaitAsync();

Timed Rate Limited Parallel Processor (e.g. Limit RPS)

Types

| Type | Source Object | Return Object | Method 1 | Method 2 | |--------------------------------------------------|---------------|---------------|--------------------| ------------------ | | TimedRateLimitedParallelAsyncProcessor | ❌ | ❌ | .WithExecutionCount(int) | .ForEachAsync(delegate) | | TimedRateLimitedParallelAsyncProcessor<TInput> | ✔ | ❌ | .WithItems(IEnumerable<TInput>) | .ForEachAsync(delegate) | | ResultTimedRateLimitedParallelAsyncProcessor<TOutput> | ❌ | ✔ | .WithExecutionCount(int) | .SelectAsync(delegate) | | ResultTimedRateLimitedParallelAsyncProcessor<TInput, TOutput> | ✔ | ✔ | .WithItems(IEnumerable<TInput>) | .SelectAsync(delegate) |

How it works
Processes your Asynchronous Tasks in Parallel, but honouring the limit that you set over the timespan that you set. As one finishes, another will start, unless you've hit the maximum allowed for the current timespan duration.

E.g. If you set a limit of 100, and a timespan of 1 second, only 100 operation should ever run at any one time over the course of a second. If the operation finishes sooner than a second (or your provided timespan), it'll wait and then start the next operation once that timespan has elapsed.

This is useful in scenarios where, for example, you have an API but it has a request per second limit

Usage

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));
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));
await voidProcessor.WaitAsync();

Caveats

• If your operations take longer than your provided TimeSpan, you probably won't get your desired throughput. This processor ensures you don't go over your rate limit, but will not increase parallel execution if you're below it.

One At A Time

Types

| Type | Source Object | Return Object | Method 1 | Method 2 | |--------------------------------------------------|---------------|---------------|--------------------| ------------------ | | OneAtATimeAsyncProcessor | ❌ | ❌ | .WithExecutionCount(int) | .ForEachAsync(delegate) | | OneAtATimeAsyncProcessor<TInput> | ✔ | ❌ | .WithItems(IEnumerable<TInput>) | .ForEachAsync(delegate) | | ResultOneAtATimeAsyncProcessor<TOutput> | ❌ | ✔ | .WithExecutionCount(int) | .SelectAsync(delegate) | | ResultOneAtATimeAsyncProcessor<TInput, TOutput> | ✔ | ✔ | .WithItems(IEnumerable<TInput>) | .SelectAsync(delegate) |

How it works
Processes your Asynchronous Tasks One at a Time. Only one will ever progress at a time. As one finishes, another will start

Usage

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)
        .ProcessOneAtATime();

// ForEachAsync for when you have nothing to return
await ids
        .ForEachAsync(id => DoSomethingAsync(id), CancellationToken.None) 
        .ProcessOneAtATime();

Caveats

• Slowest method

Batch

Types

| Type | Source Object | Return Object | Method 1 | Method 2 | |--------------------------------------------------|---------------|---------------|--------------------| ------------------ | | BatchAsyncProcessor | ❌ | ❌ | .WithExecutionCount(int) | .ForEachAsync(delegate) | | BatchAsyncProcessor<TInput> | ✔ | ❌ | .WithItems(IEnumerable<TInput>) | .ForEachAsync(delegate) | | ResultBatchAsyncProcessor<TOutput> | ❌ | ✔ | .WithExecutionCount(int) | .SelectAsync(delegate) | | ResultBatchAsyncProcessor<TInput, TOutput> | ✔ | ✔ | .WithItems(IEnumerable<TInput>) | .SelectAsync(delegate) |

How it works
Processes your Asynchronous Tasks in Batches. The next batch will not start until every Task in previous batch has finished

Usage

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)
        .ProcessInBatches(batchSize: 100);

// ForEachAsync for when you have nothing to return
await ids
        .ForEachAsync(id => DoSomethingAsync(id), CancellationToken.None) 
        .ProcessInBatches(batchSize: 100);

Caveats

• 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<TInput> | ✔ | ❌ | .WithItems(IEnumerable<TInput>) | .ForEachAsync(delegate) | | ResultParallelAsyncProcessor<TOutput> | ❌ | ✔ | .WithExecutionCount(int) | .SelectAsync(delegate) | | ResultParallelAsyncProcessor<TInput, TOutput> | ✔ | ✔ | .WithItems(IEnumerable<TInput>) | .SelectAsync(delegate) |

How it works
Processes your Asynchronous Tasks as fast as it can. All at the same time if it can

Usage

var ids = Enumerable.Range(0,