Skip to content

Parallel.ForEachAsync #533

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
SimonCropp merged 2 commits into from
Apr 10, 2026
Merged

Parallel.ForEachAsync #533

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
70e2419
Parallel.ForEachAsync
SimonCropp Apr 10, 2026
2407bbb
Merge branch 'main' into Parallel.ForEachAsync
SimonCropp Apr 10, 2026
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
2 changes: 1 addition & 1 deletion apiCount.include.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
**API count: 892**
**API count: 897**

### Per Target Framework

Expand Down
10 changes: 10 additions & 0 deletions api_list.include.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -658,6 +658,16 @@
* `bool TryGetValue<TKey, TValue>(TKey, TValue, int) where TKey : notnull` [reference](https://learn.microsoft.com/en-us/dotnet/api/system.collections.generic.ordereddictionary-2.trygetvalue?view=net-11.0#system-collections-generic-ordereddictionary-2-trygetvalue(-0-1@-system-int32@))


#### Parallel

* `Task ForEachAsync<T>(IAsyncEnumerable<T>, CancellationToken, Func<T, CancellationToken, ValueTask>)` [reference](https://learn.microsoft.com/en-us/dotnet/api/system.threading.tasks.parallel.foreachasync?view=net-11.0#system-threading-tasks-parallel-foreachasync-1(system-collections-generic-iasyncenumerable((-0))-system-threading-cancellationtoken-system-func((-0-system-threading-cancellationtoken-system-threading-tasks-valuetask))))
* `Task ForEachAsync<T>(IAsyncEnumerable<T>, Func<T, CancellationToken, ValueTask>)` [reference](https://learn.microsoft.com/en-us/dotnet/api/system.threading.tasks.parallel.foreachasync?view=net-11.0#system-threading-tasks-parallel-foreachasync-1(system-collections-generic-iasyncenumerable((-0))-system-func((-0-system-threading-cancellationtoken-system-threading-tasks-valuetask))))
* `Task ForEachAsync<T>(IAsyncEnumerable<T>, ParallelOptions, Func<T, CancellationToken, ValueTask>)` [reference](https://learn.microsoft.com/en-us/dotnet/api/system.threading.tasks.parallel.foreachasync?view=net-11.0#system-threading-tasks-parallel-foreachasync-1(system-collections-generic-iasyncenumerable((-0))-system-threading-tasks-paralleloptions-system-func((-0-system-threading-cancellationtoken-system-threading-tasks-valuetask))))
* `Task ForEachAsync<T>(IEnumerable<T>, CancellationToken, Func<T, CancellationToken, ValueTask>)` [reference](https://learn.microsoft.com/en-us/dotnet/api/system.threading.tasks.parallel.foreachasync?view=net-11.0#system-threading-tasks-parallel-foreachasync-1(system-collections-generic-ienumerable((-0))-system-threading-cancellationtoken-system-func((-0-system-threading-cancellationtoken-system-threading-tasks-valuetask))))
* `Task ForEachAsync<T>(IEnumerable<T>, Func<T, CancellationToken, ValueTask>)` [reference](https://learn.microsoft.com/en-us/dotnet/api/system.threading.tasks.parallel.foreachasync?view=net-11.0#system-threading-tasks-parallel-foreachasync-1(system-collections-generic-ienumerable((-0))-system-func((-0-system-threading-cancellationtoken-system-threading-tasks-valuetask))))
* `Task ForEachAsync<T>(IEnumerable<T>, ParallelOptions, Func<T, CancellationToken, ValueTask>)` [reference](https://learn.microsoft.com/en-us/dotnet/api/system.threading.tasks.parallel.foreachasync?view=net-11.0#system-threading-tasks-parallel-foreachasync-1(system-collections-generic-ienumerable((-0))-system-threading-tasks-paralleloptions-system-func((-0-system-threading-cancellationtoken-system-threading-tasks-valuetask))))


#### ParameterInfo

* `NullabilityState GetNullability()`
Expand Down
10 changes: 10 additions & 0 deletions readme.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1189,6 +1189,16 @@ The class `Polyfill` includes the following extension methods:
* `bool TryGetValue<TKey, TValue>(TKey, TValue, int) where TKey : notnull` [reference](https://learn.microsoft.com/en-us/dotnet/api/system.collections.generic.ordereddictionary-2.trygetvalue?view=net-11.0#system-collections-generic-ordereddictionary-2-trygetvalue(-0-1@-system-int32@))


#### Parallel

* `Task ForEachAsync<T>(IAsyncEnumerable<T>, CancellationToken, Func<T, CancellationToken, ValueTask>)` [reference](https://learn.microsoft.com/en-us/dotnet/api/system.threading.tasks.parallel.foreachasync?view=net-11.0#system-threading-tasks-parallel-foreachasync-1(system-collections-generic-iasyncenumerable((-0))-system-threading-cancellationtoken-system-func((-0-system-threading-cancellationtoken-system-threading-tasks-valuetask))))
* `Task ForEachAsync<T>(IAsyncEnumerable<T>, Func<T, CancellationToken, ValueTask>)` [reference](https://learn.microsoft.com/en-us/dotnet/api/system.threading.tasks.parallel.foreachasync?view=net-11.0#system-threading-tasks-parallel-foreachasync-1(system-collections-generic-iasyncenumerable((-0))-system-func((-0-system-threading-cancellationtoken-system-threading-tasks-valuetask))))
* `Task ForEachAsync<T>(IAsyncEnumerable<T>, ParallelOptions, Func<T, CancellationToken, ValueTask>)` [reference](https://learn.microsoft.com/en-us/dotnet/api/system.threading.tasks.parallel.foreachasync?view=net-11.0#system-threading-tasks-parallel-foreachasync-1(system-collections-generic-iasyncenumerable((-0))-system-threading-tasks-paralleloptions-system-func((-0-system-threading-cancellationtoken-system-threading-tasks-valuetask))))
* `Task ForEachAsync<T>(IEnumerable<T>, CancellationToken, Func<T, CancellationToken, ValueTask>)` [reference](https://learn.microsoft.com/en-us/dotnet/api/system.threading.tasks.parallel.foreachasync?view=net-11.0#system-threading-tasks-parallel-foreachasync-1(system-collections-generic-ienumerable((-0))-system-threading-cancellationtoken-system-func((-0-system-threading-cancellationtoken-system-threading-tasks-valuetask))))
* `Task ForEachAsync<T>(IEnumerable<T>, Func<T, CancellationToken, ValueTask>)` [reference](https://learn.microsoft.com/en-us/dotnet/api/system.threading.tasks.parallel.foreachasync?view=net-11.0#system-threading-tasks-parallel-foreachasync-1(system-collections-generic-ienumerable((-0))-system-func((-0-system-threading-cancellationtoken-system-threading-tasks-valuetask))))
* `Task ForEachAsync<T>(IEnumerable<T>, ParallelOptions, Func<T, CancellationToken, ValueTask>)` [reference](https://learn.microsoft.com/en-us/dotnet/api/system.threading.tasks.parallel.foreachasync?view=net-11.0#system-threading-tasks-parallel-foreachasync-1(system-collections-generic-ienumerable((-0))-system-threading-tasks-paralleloptions-system-func((-0-system-threading-cancellationtoken-system-threading-tasks-valuetask))))


#### ParameterInfo

* `NullabilityState GetNullability()`
Expand Down
10 changes: 10 additions & 0 deletions src/Consume/Consume.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1409,6 +1409,16 @@ async Task Task_WhenEach_Methods()
}
#endif

#if FeatureValueTask
async Task Parallel_Methods()
{
var items = new[] { 1, 2, 3 };
await Parallel.ForEachAsync(items, (_, _) => default);
await Parallel.ForEachAsync(items, CancellationToken.None, (_, _) => default);
await Parallel.ForEachAsync(items, new ParallelOptions(), (_, _) => default);
}
#endif

void TaskCompletionSource_NonGeneric_Methods()
{
var tcs = new TaskCompletionSource();
Expand Down
145 changes: 145 additions & 0 deletions src/Polyfill/ParallelPolyfill.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,145 @@
#if !NET6_0_OR_GREATER && FeatureValueTask

namespace Polyfills;

using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;

static partial class Polyfill
{
extension(Parallel)
{
/// <summary>
/// Executes a for-each operation on an <see cref="IEnumerable{T}"/> in which iterations may run in parallel.
/// </summary>
//Link: https://learn.microsoft.com/en-us/dotnet/api/system.threading.tasks.parallel.foreachasync?view=net-11.0#system-threading-tasks-parallel-foreachasync-1(system-collections-generic-ienumerable((-0))-system-threading-tasks-paralleloptions-system-func((-0-system-threading-cancellationtoken-system-threading-tasks-valuetask)))
public static async Task ForEachAsync<T>(
IEnumerable<T> source,
ParallelOptions parallelOptions,
Func<T, CancellationToken, ValueTask> body)
{
using var semaphore = new SemaphoreSlim(
parallelOptions.MaxDegreeOfParallelism > 0
? parallelOptions.MaxDegreeOfParallelism
: Environment.ProcessorCount);

var tasks = new List<Task>();
foreach (var item in source)
{
tasks.Add(
Task.Factory.StartNew(
async () =>
{
await semaphore
.WaitAsync(parallelOptions.CancellationToken)
.ConfigureAwait(false);
try
{
await body(item, parallelOptions.CancellationToken)
.ConfigureAwait(false);
}
finally
{
semaphore.Release();
}
},
parallelOptions.CancellationToken,
TaskCreationOptions.None,
parallelOptions.TaskScheduler ?? TaskScheduler.Default)
.Unwrap());
}

await Task.WhenAll(tasks).ConfigureAwait(false);
}

/// <summary>
/// Executes a for-each operation on an <see cref="IEnumerable{T}"/> in which iterations may run in parallel.
/// </summary>
//Link: https://learn.microsoft.com/en-us/dotnet/api/system.threading.tasks.parallel.foreachasync?view=net-11.0#system-threading-tasks-parallel-foreachasync-1(system-collections-generic-ienumerable((-0))-system-threading-cancellationtoken-system-func((-0-system-threading-cancellationtoken-system-threading-tasks-valuetask)))
public static Task ForEachAsync<T>(
IEnumerable<T> source,
CancellationToken cancellationToken,
Func<T, CancellationToken, ValueTask> body) =>
Parallel.ForEachAsync(source, new ParallelOptions { CancellationToken = cancellationToken }, body);

/// <summary>
/// Executes a for-each operation on an <see cref="IEnumerable{T}"/> in which iterations may run in parallel.
/// </summary>
//Link: https://learn.microsoft.com/en-us/dotnet/api/system.threading.tasks.parallel.foreachasync?view=net-11.0#system-threading-tasks-parallel-foreachasync-1(system-collections-generic-ienumerable((-0))-system-func((-0-system-threading-cancellationtoken-system-threading-tasks-valuetask)))
public static Task ForEachAsync<T>(
IEnumerable<T> source,
Func<T, CancellationToken, ValueTask> body) =>
Parallel.ForEachAsync(source, CancellationToken.None, body);

#if FeatureAsyncInterfaces

/// <summary>
/// Executes a for-each operation on an <see cref="IAsyncEnumerable{T}"/> in which iterations may run in parallel.
/// </summary>
//Link: https://learn.microsoft.com/en-us/dotnet/api/system.threading.tasks.parallel.foreachasync?view=net-11.0#system-threading-tasks-parallel-foreachasync-1(system-collections-generic-iasyncenumerable((-0))-system-threading-tasks-paralleloptions-system-func((-0-system-threading-cancellationtoken-system-threading-tasks-valuetask)))
public static async Task ForEachAsync<T>(
IAsyncEnumerable<T> source,
ParallelOptions parallelOptions,
Func<T, CancellationToken, ValueTask> body)
{
using var semaphore = new SemaphoreSlim(
parallelOptions.MaxDegreeOfParallelism > 0
? parallelOptions.MaxDegreeOfParallelism
: Environment.ProcessorCount);

var tasks = new List<Task>();

await foreach (var item in source.WithCancellation(parallelOptions.CancellationToken))
{
tasks.Add(
Task.Factory.StartNew(
async () =>
{
await semaphore
.WaitAsync(parallelOptions.CancellationToken)
.ConfigureAwait(false);
try
{
await body(item, parallelOptions.CancellationToken)
.ConfigureAwait(false);
}
finally
{
semaphore.Release();
}
},
parallelOptions.CancellationToken,
TaskCreationOptions.None,
parallelOptions.TaskScheduler ?? TaskScheduler.Default)
.Unwrap());
}

await Task.WhenAll(tasks).ConfigureAwait(false);
}

/// <summary>
/// Executes a for-each operation on an <see cref="IAsyncEnumerable{T}"/> in which iterations may run in parallel.
/// </summary>
//Link: https://learn.microsoft.com/en-us/dotnet/api/system.threading.tasks.parallel.foreachasync?view=net-11.0#system-threading-tasks-parallel-foreachasync-1(system-collections-generic-iasyncenumerable((-0))-system-threading-cancellationtoken-system-func((-0-system-threading-cancellationtoken-system-threading-tasks-valuetask)))
public static Task ForEachAsync<T>(
IAsyncEnumerable<T> source,
CancellationToken cancellationToken,
Func<T, CancellationToken, ValueTask> body) =>
Parallel.ForEachAsync(source, new ParallelOptions { CancellationToken = cancellationToken }, body);

/// <summary>
/// Executes a for-each operation on an <see cref="IAsyncEnumerable{T}"/> in which iterations may run in parallel.
/// </summary>
//Link: https://learn.microsoft.com/en-us/dotnet/api/system.threading.tasks.parallel.foreachasync?view=net-11.0#system-threading-tasks-parallel-foreachasync-1(system-collections-generic-iasyncenumerable((-0))-system-func((-0-system-threading-cancellationtoken-system-threading-tasks-valuetask)))
public static Task ForEachAsync<T>(
IAsyncEnumerable<T> source,
Func<T, CancellationToken, ValueTask> body) =>
Parallel.ForEachAsync(source, CancellationToken.None, body);

#endif
}
}

#endif
125 changes: 125 additions & 0 deletions src/Split/net461/ParallelPolyfill.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,125 @@
// <auto-generated />
#pragma warning disable
#if FeatureValueTask
namespace Polyfills;
using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
static partial class Polyfill
{
extension(Parallel)
{
/// <summary>
/// Executes a for-each operation on an <see cref="IEnumerable{T}"/> in which iterations may run in parallel.
/// </summary>
public static async Task ForEachAsync<T>(
IEnumerable<T> source,
ParallelOptions parallelOptions,
Func<T, CancellationToken, ValueTask> body)
{
using var semaphore = new SemaphoreSlim(
parallelOptions.MaxDegreeOfParallelism > 0
? parallelOptions.MaxDegreeOfParallelism
: Environment.ProcessorCount);
var tasks = new List<Task>();
foreach (var item in source)
{
tasks.Add(
Task.Factory.StartNew(
async () =>
{
await semaphore
.WaitAsync(parallelOptions.CancellationToken)
.ConfigureAwait(false);
try
{
await body(item, parallelOptions.CancellationToken)
.ConfigureAwait(false);
}
finally
{
semaphore.Release();
}
},
parallelOptions.CancellationToken,
TaskCreationOptions.None,
parallelOptions.TaskScheduler ?? TaskScheduler.Default)
.Unwrap());
}
await Task.WhenAll(tasks).ConfigureAwait(false);
}
/// <summary>
/// Executes a for-each operation on an <see cref="IEnumerable{T}"/> in which iterations may run in parallel.
/// </summary>
public static Task ForEachAsync<T>(
IEnumerable<T> source,
CancellationToken cancellationToken,
Func<T, CancellationToken, ValueTask> body) =>
Parallel.ForEachAsync(source, new ParallelOptions { CancellationToken = cancellationToken }, body);
/// <summary>
/// Executes a for-each operation on an <see cref="IEnumerable{T}"/> in which iterations may run in parallel.
/// </summary>
public static Task ForEachAsync<T>(
IEnumerable<T> source,
Func<T, CancellationToken, ValueTask> body) =>
Parallel.ForEachAsync(source, CancellationToken.None, body);
#if FeatureAsyncInterfaces
/// <summary>
/// Executes a for-each operation on an <see cref="IAsyncEnumerable{T}"/> in which iterations may run in parallel.
/// </summary>
public static async Task ForEachAsync<T>(
IAsyncEnumerable<T> source,
ParallelOptions parallelOptions,
Func<T, CancellationToken, ValueTask> body)
{
using var semaphore = new SemaphoreSlim(
parallelOptions.MaxDegreeOfParallelism > 0
? parallelOptions.MaxDegreeOfParallelism
: Environment.ProcessorCount);
var tasks = new List<Task>();
await foreach (var item in source.WithCancellation(parallelOptions.CancellationToken))
{
tasks.Add(
Task.Factory.StartNew(
async () =>
{
await semaphore
.WaitAsync(parallelOptions.CancellationToken)
.ConfigureAwait(false);
try
{
await body(item, parallelOptions.CancellationToken)
.ConfigureAwait(false);
}
finally
{
semaphore.Release();
}
},
parallelOptions.CancellationToken,
TaskCreationOptions.None,
parallelOptions.TaskScheduler ?? TaskScheduler.Default)
.Unwrap());
}
await Task.WhenAll(tasks).ConfigureAwait(false);
}
/// <summary>
/// Executes a for-each operation on an <see cref="IAsyncEnumerable{T}"/> in which iterations may run in parallel.
/// </summary>
public static Task ForEachAsync<T>(
IAsyncEnumerable<T> source,
CancellationToken cancellationToken,
Func<T, CancellationToken, ValueTask> body) =>
Parallel.ForEachAsync(source, new ParallelOptions { CancellationToken = cancellationToken }, body);
/// <summary>
/// Executes a for-each operation on an <see cref="IAsyncEnumerable{T}"/> in which iterations may run in parallel.
/// </summary>
public static Task ForEachAsync<T>(
IAsyncEnumerable<T> source,
Func<T, CancellationToken, ValueTask> body) =>
Parallel.ForEachAsync(source, CancellationToken.None, body);
#endif
}
}
#endif
Loading
Loading