Throttled execution of an enumeration of Tasks

Question

Example of use:

var users = await usernames.SelectTaskResults(GetUserDetails, 4);

where GetUserDetails(string username) is a method that calls HttpClient to access an API and returns a User object. The SelectTaskResults returns the awaited tasks. At most, there will be 4 concurrent threads calling the API at the same time for this example.

Concerns: I'm worried .Wait() could cause a deadlock.

Code:

public static async Task<IEnumerable<TResult>> SelectTaskResults<TInput, TResult>(
        this IEnumerable<TInput> source,
        Func<TInput, Task<TResult>> taskFunc,
        int degreesOfParallelism = 1,
        bool throwFaulted = false,
        CancellationToken cancellationToken = default(CancellationToken))
    {
        // Task.Run creates the task but it doesn't start executing immediately - for debugging
        var tasks = source
            .Select(input => Task.Run(() => taskFunc(input), cancellationToken))
            .ToArray();
        
        Parallel.ForEach(tasks,
            new ParallelOptions { MaxDegreeOfParallelism = degreesOfParallelism }
            , task =>
            {
                try
                {
                    task.Wait(cancellationToken); // .Start() doesn't work for promise-like tasks
                }
                catch (Exception)
                {
                    if (throwFaulted)
                    {
                        throw;
                    }
                }
            }
        );
        
        var output = await Task.WhenAll(tasks.Where(x=> !x.IsFaulted));
        
        return output.AsEnumerable();
    }

Answer 1

A TPL inside async implementation: mixing of implementaion concepts makes the code complicated or even buggy. Finally Parallel.ForEach is blocking the current thread which isn't allowed in async programming.

Also avoid blocking operations on Task e.g. .Wait() or .Result for not completed Task.

But I suggest totally different approach. Consider the simplified example.

public class ThrottledApiClient : IDisposable
{
    private readonly HttpClient _client;
    private readonly SemaphoreSlim _semaphore;

    public ThrottledApiClient(string baseUrl, int concurrentDegree = 0)
    {
        _client = new HttpClient();
        _client.BaseAddress = new Uri(baseUrl);
        _semaphore = new SemaphoreSlim(concurrentDegree == 0 ? Environment.ProcessorCount * 2 : concurrentDegree);
    }

    public async Task<string[]> PostJsonRequestsAsync(IEnumerable<(string, string)> requests, CancellationToken token)
    {
        List<Task<string>> tasks = new List<Task<string>>();
        // wrap the loop with try-catch (OperationCanceledException) to handle Cancellation right here if necessary
        foreach ((string path, string json) in requests)
        {
            await _semaphore.WaitAsync(token);
            tasks.Add(PostJsonAsync(path, json, token));
        }

        return await Task.WhenAll(tasks);

        // or with filter
        //await Task.WhenAll(tasks);
        //return tasks.Where(t => t.IsCompletedSucessfully).Select(t => t.Result).ToArray();
    }

    private async Task<string> PostJsonAsync(string url, string json, CancellationToken token)
    {
        try
        {
            using HttpContent content = new StringContent(json, Encoding.UTF8, "application/json");
            using HttpResponseMessage response = await _client.PostAsync(url, content, token);
            // response.EnsureSuccessStatusCode(); // to throw on not success instead of returning result
            return await response.Content.ReadAsStringAsync(token);
        }
        // catch (HttpRequestException) 
        // catch (OperationCanceledException) { return null; } // to swallow cancellation
        finally
        {
            _semaphore.Release();
        }
    }

    public void Dispose()
    {
        _client.Dispose();
        _semaphore.Dispose();
    }
}

And the demo.

(string, string)[] requests = new (string, string)[]
{
    ("/getAccounts", "{...}"),
    ("/getMediaPath", "{...}")
};

using (var api = new ThrottledApiClient("https://api.example.com", 4))
using (var cts = new CancellationTokenSource())
{
    var results = await api.PostJsonRequestsAsync(requests, cts.Token);
    //...
}

You can invoke PostJsonRequestsAsync even concurrently multiple times, even from different threads but only 4 requests will be active simoultaneously.

SemaphoreSlim is default synchronization tool in async programming for now.

Answer 2

thanks to @aepot for helping me clean this up.

Changes from original:

Tasks now start immediately when being added to tasks list to bypass the need to start them.

SemaphoreSlim is used over Parallel.ForEach to control MaxDegreeOfParallelism as to not not deadlock.

tasks.Where(x=> !x.IsFaulted) is removed because its sort of a race-condition where a task could Fault between the Where and WhenAll.

throwFaulted was replaced with the defaultForException Func so if there is an exception, we can give it a replacement value .

CancellationToken is now supported.

Extention Method was renamed to InvokeFunctionsAsync

My current solution:

var users = await usernames.InvokeFunctionsAsync(GetUserDetails, 4, username=> new User
                {
                    Username = username,
                    Domain = 'US'
                });

    public static async Task<IEnumerable<TResult>> InvokeFunctionsAsync<TInput, TResult>(
        this IEnumerable<TInput> source,
        Func<TInput, Task<TResult>> task,
        int degreesOfParallelism = 1,
        Func<TInput, TResult> defaultForException = null,
        CancellationToken cancellationToken = default)
    {
        using var throttle = new SemaphoreSlim(degreesOfParallelism);
        var tasks = new List<Task<TResult>>();

        foreach (var input in source)
        {
            await throttle.WaitAsync(cancellationToken);

            tasks.Add(Run(input, throttle));
        }
        
        return await Task.WhenAll(tasks);

        async Task<TResult> Run(TInput value, SemaphoreSlim semaphore)
        {
            try
            {
                return await task(value);
            }
            catch (Exception)
            {
                if (defaultForException == null)
                {
                    throw;
                }
                return defaultForException(value);
            }
            finally
            {
                semaphore.Release();
            }
        }
    }

Answer 3

I would suggest you use something like System.Threading.Tasks.Dataflow or System.Reactive from NuGet.

Dataflow in particular is made to feed data through a pipeline.

• Consider passing the CancellationToken to the delegate.
• Always validate arguments
• Smaller more specializes methods can make the code easier to understand.

Example:

public static Task<IEnumerable<TResult>> SelectTaskResults<TInput, TResult>(
    this IEnumerable<TInput> source,
    Func<TInput, Task<TResult>> task,
    int degreesOfParallelism = 1,
    Func<TInput, TResult> defaultForException = null,
    CancellationToken cancellationToken = default)
{
    if (source is null) throw new ArgumentNullException(nameof(source));
    if (task is null) throw new ArgumentNullException(nameof(task));
    if (degreesOfParallelism < 1 && degreesOfParallelism != -1) throw new ArgumentOutOfRangeException(nameof(degreesOfParallelism));
    
    return SelectTaskResults(source, (i, c) => task(i), degreesOfParallelism, defaultForException, cancellationToken);
}

public static async Task<IEnumerable<TResult>> SelectTaskResults<TInput, TResult>(
    this IEnumerable<TInput> source,
    Func<TInput, CancellationToken, Task<TResult>> task,
    int degreesOfParallelism = 1,
    Func<TInput, TResult> defaultForException = null,
    CancellationToken cancellationToken = default)
{
    if (source is null) throw new ArgumentNullException(nameof(source));
    if (task is null) throw new ArgumentNullException(nameof(task));
    if (degreesOfParallelism < 1 && degreesOfParallelism != -1) throw new ArgumentOutOfRangeException(nameof(degreesOfParallelism));
    
    var throttleOptions = new ExecutionDataflowBlockOptions
    {
        MaxDegreeOfParallelism = degreesOfParallelism,
        CancellationToken = cancellationToken,
        EnsureOrdered = true,
        BoundedCapacity = Math.Max(degreesOfParallelism * 2, 4),
    };

    var executionBlock = new TransformBlock<TInput, TResult>(
        input => ProcessInput(input, task, defaultForException),
        throttleOptions
    );

    var resultTask = executionBlock.ToListAsync(cancellationToken);

    foreach (var element in source)
    {
        while (!await executionBlock.SendAsync(element, cancellationToken).ConfigureAwait(false)) ;
    }

    executionBlock.Complete();

    return await resultTask;
}

public static async Task<List<T>> ToListAsync<T>(this IReceivableSourceBlock<T> block, CancellationToken cancellationToken = default)
{
    var list = new List<T>();
    while (await block.OutputAvailableAsync(cancellationToken).ConfigureAwait(false))
    {
        while (block.TryReceive(out var item))
        {
            list.Add(item);
        }
    }
    await block.Completion.ConfigureAwait(false); // Propagate possible exception
    return list;
}

private static async Task<TResult> ProcessInput<TInput, TResult>(TInput input, Func<TInput, CancellationToken, Task<TResult>> process, Func<TInput, TResult> defaultForException = null, CancellationToken cancellationToken = default)
{
    try
    {
        return await process(input, cancellationToken).ConfigureAwait(false);
    }
    catch (TaskCanceledException)
    {
        throw;
    }
    catch (Exception) when (defaultForException != null)
    {
        return defaultForException(input);
    }
}

With Dataflow you can tweak how the parallelism and order of how the tasks are processed by tweaking the options.