Simply the concurrency limit with task

[WeihanLi]

Background and Motivation

Currently, if we need to limit the concurrency, it may be a little hard to implement(maybe there's something I missed)

When we use Parallel, we could use MaxDegreeOfParallelism to limit the max concurrency, if we could add something like that to Task.WhenAll and Task.WhenAny method

When I use Task.WhenAll, I usually use a SemephoreSlim to limit the concurrency like follows:

var nums = Enmuerable.Range(1, 1000);
using var semaphore = new SemaphoreSlim(1, 50);
var result = await Task.WhenAll(
    nums
        .Select(async x =>
        {
            try
            {
                await semaphore.WaitAsync();
                return await GetDetail(x);
            }
            finally
            {
                semaphore.Release();
            }
        })
).ToArray());

I think it's maybe a common requirement, hope there's a simple API to implement this.

Proposed API

namespace System.Threading.Tasks
{
     public class Task{
+      public static Task WhenAll(IEnumerable<Task> tasks, int maxDegreeOfParallelism);
+      public static Task<TResult[]> WhenAll<TResult>(IEnumerable<Task<TResult>> tasks, int maxDegreeOfParallelism);
+      public static Task<Task> WhenAny(IEnumerable<Task> tasks, int maxDegreeOfParallelism);
+      public static Task<Task<TResult>> WhenAny<TResult>(IEnumerable<Task<TResult>> tasks, int maxDegreeOfParallelism);
     }

Usage Examples

var nums = Enmuerable.Range(1, 1000);
await Task.WhenAll(nums.Select(async n => {
   // async code block
}), 50);

Risks

Maybe there's something I missed.

Tagging subscribers to this area: @carlossanlop
See info in area-owners.md if you want to be subscribed.

Issue Details

---

Background and Motivation

Currently, if we need to limit the concurrency, it may be a little hard to implement(maybe there's something I missed)

When we use Parallel, we could use MaxDegreeOfParallelism to limit the max concurrency, if we could add something like that to Task.WhenAll and Task.WhenAny method

When I use Task.WhenAll, I usually use a SemephoreSlim to limit the concurrency like follows:

var nums = Enmuerable.Range(1, 1000);
using var semaphore = new SemaphoreSlim(1, 50);
var result = await Task.WhenAll(
    nums
        .Select(async x =>
        {
            try
            {
                await semaphore.WaitAsync();
                return await GetDetail(x);
            }
            finally
            {
                semaphore.Release();
            }
        })
).ToArray());

I think it's maybe a common requirement, hope there's a simple API to implement this.

Proposed API

namespace System.Threading.Tasks
{
     public class Task{
+      public static Task WhenAll(IEnumerable<Task> tasks, int maxDegreeOfParallelism);
+      public static Task<TResult[]> WhenAll<TResult>(IEnumerable<Task<TResult>> tasks, int maxDegreeOfParallelism);
+      public static Task<Task> WhenAny(IEnumerable<Task> tasks, int maxDegreeOfParallelism);
+      public static Task<Task<TResult>> WhenAny<TResult>(IEnumerable<Task<TResult>> tasks, int maxDegreeOfParallelism);
     }

Usage Examples

var nums = Enmuerable.Range(1, 1000);
await Task.WhenAll(nums.Select(async n => {
   // async code block
}).ToArray(), 50);

Risks

Maybe there's something I missed.

Author:	WeihanLi
Assignees:	-
Labels:	api-suggestion, area-System.Threading.Tasks, untriaged
Milestone:	-

[huoyaoyuan]

When a Task is passed to WhenAll, it is already running and may take a thread. Any limitation may only apply when it awaits something.
An api with custom TaskSchedular makes more sense.

[stephentoub]

Sounds like you're asking for Parallel.ForEachAsync, which already exists for .NET 6 and supports ParallelOptions, including both TaskScheduler and MaxDegreeOfParallelism.

runtime/src/libraries/System.Threading.Tasks.Parallel/ref/System.Threading.Tasks.Parallel.cs

Lines 39 to 44 in 911640b

	public static System.Threading.Tasks.Task ForEachAsync<TSource>(System.Collections.Generic.IEnumerable<TSource> source, System.Func<TSource, CancellationToken, ValueTask> body) { throw null; }
	public static System.Threading.Tasks.Task ForEachAsync<TSource>(System.Collections.Generic.IEnumerable<TSource> source, CancellationToken cancellationToken, System.Func<TSource, CancellationToken, ValueTask> body) { throw null; }
	public static System.Threading.Tasks.Task ForEachAsync<TSource>(System.Collections.Generic.IEnumerable<TSource> source, System.Threading.Tasks.ParallelOptions parallelOptions, System.Func<TSource, CancellationToken, ValueTask> body) { throw null; }
	public static System.Threading.Tasks.Task ForEachAsync<TSource>(System.Collections.Generic.IAsyncEnumerable<TSource> source, System.Func<TSource, CancellationToken, ValueTask> body) { throw null; }
	public static System.Threading.Tasks.Task ForEachAsync<TSource>(System.Collections.Generic.IAsyncEnumerable<TSource> source, CancellationToken cancellationToken, System.Func<TSource, CancellationToken, ValueTask> body) { throw null; }
	public static System.Threading.Tasks.Task ForEachAsync<TSource>(System.Collections.Generic.IAsyncEnumerable<TSource> source, System.Threading.Tasks.ParallelOptions parallelOptions, System.Func<TSource, CancellationToken, ValueTask> body) { throw null; }

[WeihanLi]

Parallel.ForEachAsync is exactly what I need, thanks greatly for your work @stephentoub

[stephentoub]

Great 😄