3
\$\begingroup\$

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();
    }
\$\endgroup\$
1
  • 4
    \$\begingroup\$ Welcome to Code Review! I rolled back your last edit. After getting an answer you are not allowed to change your code anymore. This is to ensure that answers do not get invalidated and have to hit a moving target. If you have changed your code you can either post it as an answer (if it would constitute a code review) or ask a new question with your changed code (linking back to this one as reference). See the section What should I not do? on What should I do when someone answers my question? for more information \$\endgroup\$ Apr 7 at 15:29
4
\$\begingroup\$

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.

\$\endgroup\$
9
  • \$\begingroup\$ Thanks for the response. SemaphoreSlim is what I tried first. SemaphoreSlim first argument sets the initial concurrentDegree and the 2nd sets the max. When I set the max, I got an exception that said SemaphoreSlim exceed it's max. \$\endgroup\$ Apr 7 at 13:28
  • 1
    \$\begingroup\$ @PaulTotzke You have no reason to trust me, correct. But first argument sets the initial concurrentDegree and the 2nd sets the max = nope. Single argument means "you have 4 free slots", two arguments means "you have 4 slots and initially 2 of them busy", where busy slots - first argument, total degree - second. When you have all slots free and call .Release(), you get an exception. Use one argument for constructor. \$\endgroup\$
    – aepot
    Apr 7 at 13:33
  • 1
    \$\begingroup\$ @PaulTotzke a fix: public SemaphoreSlim (int initialCount, int maxCount) means new SemaphoreSlim(freeSlots, maxSlots) \$\endgroup\$
    – aepot
    Apr 7 at 14:15
  • \$\begingroup\$ I'm testing out SemaphoreSlim again. This exception turned me off using it but I might have misunderstood: System.Threading.SemaphoreFullException: Adding the specified count to the semaphore would cause it to exceed its maximum count. at System.Threading.SemaphoreSlim.Release(Int32 releaseCount) at System.Threading.SemaphoreSlim.Release() \$\endgroup\$ Apr 7 at 15:17
  • 1
    \$\begingroup\$ Let us continue this discussion in chat. \$\endgroup\$ Apr 7 at 15:33
2
\$\begingroup\$

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();
            }
        }
    }
\$\endgroup\$
4
  • \$\begingroup\$ Welcome to Code Review! While this is an alternate solution it doesn't make a good review. please explain what you did and why you did it. as is this answer may receive downvotes and/or be deleted. \$\endgroup\$
    – Malachi
    Apr 7 at 15:35
  • \$\begingroup\$ @Malachi this is OP's post as per If you have changed your code you can either post it as an answer in the comment below the question. \$\endgroup\$
    – aepot
    Apr 7 at 15:38
  • 2
    \$\begingroup\$ @aepot All answers should still be reviews. It's perfectly fine to upload a new solution as part of that. If the answer at least tells what's different in regards to the original and a synopsis of why it's better, that would be perfectly fine. \$\endgroup\$
    – Mast
    Apr 7 at 17:16
  • 1
    \$\begingroup\$ I hope this is better. \$\endgroup\$ Apr 7 at 22:59
1
\$\begingroup\$

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.

\$\endgroup\$
1
  • \$\begingroup\$ await block.OutputAvailableAsync(cancellationToken).ConfigureAwait(false) can it be completed synchronously? \$\endgroup\$
    – aepot
    Apr 12 at 18:27

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.