Parallel and SemaphoreSlim
Parallel.ForEachAsync fans work out over many items, and SemaphoreSlim caps how much runs at once so you do not flood your own database.
await Parallel.ForEachAsync(items, new ParallelOptions { MaxDegreeOfParallelism = 4 }, async (item, token) => { /* body */ });Parallel.ForEachAsync runs an asynchronous body once for every item in a sequence, several items in flight at a time, and returns a Task that completes when all of them are done. It was added in .NET 6. Unlike the older Parallel.ForEach, it understands await, so no thread sits blocked while a database or an HTTP call answers.
Capping the fan-out
By default it runs at most Environment.ProcessorCount bodies at once. Pass a ParallelOptions with MaxDegreeOfParallelism to choose the number yourself, and set its CancellationToken so it stops handing out new items when the work is abandoned.
Capping it is not politeness. Five thousand rows fanned out at once means five thousand database connections asked of a pool that holds a hundred. The rest queue, then time out, and your request handlers start failing while the database itself is barely busy. Unbounded parallelism is a denial of service you perform on yourself.
SemaphoreSlim, the longhand
A SemaphoreSlim holds a number of permits. WaitAsync takes one, waiting without blocking a thread when none is free, and Release hands one back. new SemaphoreSlim(4) plus Task.WhenAll is the same cap built by hand. Use it when the items are not a plain sequence, or when you need each result returned. One shared semaphore caps a whole service; a new one created per request caps nothing.
Whatever the body touches is touched concurrently, so shared state inside it still needs a lock or an atomic operation.
Example
using System.Collections.Concurrent;
int[] ids = [1, 2, 3, 4, 5, 6];
ConcurrentBag<int> squares = new();
ParallelOptions options = new()
{
MaxDegreeOfParallelism = 3 // at most three bodies in flight at once
};
await Parallel.ForEachAsync(ids, options, async (id, token) =>
{
await Task.Delay(10, token); // stands in for a database or HTTP call
squares.Add(id * id); // ConcurrentBag is safe to add to from many threads
});
// Bodies finish in whatever order they finish, so sort before printing.
Console.WriteLine(string.Join(", ", squares.Order()));
// Output:
// 1, 4, 9, 16, 25, 36When to use it
- Calling an HTTP API for five thousand records without opening five thousand connections.
- Rebuilding a search index from a database with four rows in flight at a time.
- Capping outbound calls to a rate-limited third-party service across the whole process.
- Hashing a large batch of files across all available cores.
More examples
The same cap, shorthand and longhand
Both methods send every id with at most four calls outstanding. Prefer the short one. Write the long one when the shape does not fit: when you need the returned values, when items arrive from somewhere that is not an IEnumerable, or when different items need different permits. Note that every task is created before the first await on the semaphore, so the permits control the work, not the task objects.
using System.Threading;
internal static class Sender
{
// Shorthand: at most four bodies at a time, cancellation included.
public static Task SendAllShortAsync(int[] ids, CancellationToken token)
{
ParallelOptions options = new()
{
MaxDegreeOfParallelism = 4,
CancellationToken = token
};
return Parallel.ForEachAsync(ids, options, SendOneAsync);
}
// Longhand: the same cap by hand. Four permits, one taken per item.
public static async Task SendAllLongAsync(int[] ids, CancellationToken token)
{
using SemaphoreSlim gate = new(4);
List<Task> running = new();
foreach (int id in ids)
{
running.Add(SendThroughGateAsync(id));
}
await Task.WhenAll(running);
async Task SendThroughGateAsync(int value)
{
await gate.WaitAsync(token);
try
{
await SendOneAsync(value, token);
}
finally
{
gate.Release(); // in a finally, so a throw still frees the permit
}
}
}
private static async ValueTask SendOneAsync(int id, CancellationToken token)
{
await Task.Delay(10, token);
}
}One gate for the whole service
Reach for this when the thing you are protecting is shared by the whole process, such as a connection pool or a rate-limited third-party API. The common mistake is to cap each operation rather than the resource: a hundred concurrent requests each politely limiting themselves to eight still ask for eight hundred connections. Set the number from what the resource can take, not from what feels generous.
using System.Threading;
internal sealed class ReportService
{
// Static, so every caller in the process shares these eight permits.
// A SemaphoreSlim created inside the method would cap nothing, because
// each call would get its own fresh eight.
private static readonly SemaphoreSlim DatabaseGate = new(8);
public async Task<int> CountRowsAsync(string table, CancellationToken token)
{
await DatabaseGate.WaitAsync(token);
try
{
await Task.Delay(5, token); // stands in for the query
return table.Length;
}
finally
{
DatabaseGate.Release();
}
}
}Parallel.For for CPU-bound work
Use Parallel.For and Parallel.ForEach for work that keeps a processor busy: hashing, resizing images, scoring rows already in memory. The distinction that matters is not how long the work takes but whether it is waiting or computing. Calling a blocking method inside Parallel.ForEach is the most common way to turn a fast service into one that times out under load.
using System.Collections.Concurrent;
ConcurrentBag<int> squares = new();
// Parallel.For is the synchronous form: it splits the range across pool
// threads and returns when every body has run. There is no await inside,
// so occupying a thread is exactly the point.
Parallel.For(1, 5, i => squares.Add(i * i));
Console.WriteLine(string.Join(", ", squares.Order()));
// Output:
// 1, 4, 9, 16
// For work that waits - a query, an HTTP call - use Parallel.ForEachAsync.
// Parallel.For would hold a thread idle while the answer travelled, and the
// thread pool would grow one slow thread at a time to compensate.
Discussion