SemaphoreSlim limit tasks

Question

I find myself having to loop through a lot of async calls all over the place in my code base. The current pattern we are using is to declare SemaphoreSlim(maxcount) and then await sem.WaitAsync(), Create a new task, add it to a List and then repeat. To control the release the new task itself has a reference to sem and does the release on final. There is a step in there to look for and remove completed tasks. This is a pattern that was inherited and is used for multiple different types of async calls. As I develop new code I was hoping to simply this into a single helper class where I can just queue the work until I hit a set limit and then have to wait to add the next once a slot as freed up.

The calls are all async (as this is a 3rd party lib that we don't control). Also, the library I maintain has to be used in both winform and asp.net processes, so keeping it async seems ideal. The targets are usually waiting web services (5-10s range), in which we don't want to slam, but at the same time we typically have 1000's of items in the queue (3rd part has no bulk update implementation -- one element at a time type situation).

This is what I have come up with (naming is still a work in progress):

// Assume this is the parameter to the method we're calling over and over again
private class _ProcessArg
{
  public Guid ID { get; set; }
}

// The method that we are calling over and over again
private async Task _Process(_ProcessArg arg, CancellationToken ct)
{
  await Task.Run(() => { System.Diagnostics.Debug.WriteLine(arg.ID.ToString())});
}

// The main loop where we generated the data for the call. 
private async Task _RunMainLoop(CancellationToken ct)
{
  int maxThreads = 10;
  ConcurrentQueue<Guid> queue = new ConcurrentQueue<Guid>();

  // Typically this would be the database load/whatnot
  for (int i = 0; i < 100000; i++)
  {
    queue.Enqueue(Guid.NewGuid());
  }

  AsyncTaskMutex mutex = new AsyncTaskMutex(maxThreads);
  while (true)
  {
    Guid id;
    if (queue.TryDequeue(out id))
    {
      await mutex.QueueTask<_ProcessArg>(_Process, new _ProcessArg
      {
        ID = id,
      }, ct);
    }
    else
    {
      await mutex.DrainQueue(ct);
      break;
    }
  }
}


// Class I'm looking for peer review on
public class AsyncTaskMutex
{
  private SemaphoreSlim _sem;
  private List<Task> _tasks;

  public AsyncTaskMutex()
    : this(10)
  {
  }

  public AsyncTaskMutex(int maxTasks)
  {
    _sem = new SemaphoreSlim(maxTasks, maxTasks);
    _tasks = new List<Task>();
  }

  public async Task DrainQueue(CancellationToken ct)
  {
    await Task.WhenAll(_tasks);
    _tasks.RemoveAll(t => t.IsCompleted);
  }

  public async Task QueueTask<T>(Func<T, CancellationToken, Task> func, T args, CancellationToken ct = default(CancellationToken))
  {
    await _sem.WaitAsync(ct);
    try
    {
      Task task = func(args, ct);
      task.GetAwaiter().OnCompleted(_OnCompleted);
      _tasks.Add(task);
    }
    catch (OperationCanceledException)
    {
      // Intentional ignore
      return;
    }
  }

  private void _OnCompleted()
  {
    _sem.Release(1);
    _tasks.RemoveAll(t => t.IsCompleted);
  }

}

Answer 1

It seems you simply need process a fixed set of work items in parallel with a fixed degree of parallelism and in an async compatible way. Stephen Toub has written a very elegant way to do that in just a couple lines of code.

public static Task ForEachAsync<T>(this IEnumerable<T> source, int dop, Func<T, Task> body) 
{ 
    return Task.WhenAll( 
        from partition in Partitioner.Create(source).GetPartitions(dop) 
        select Task.Run(async delegate { 
            using (partition) 
                while (partition.MoveNext()) 
                    await body(partition.Current); 
        })); 
}

Your new code would be:

await ForEachAsync(GetWorkItems(), dop: 16, body: async item => {
 await ProcessItem(item); //TODO
});

There is no need to explicitly maintain a queue. But if you want to, you can du that by feeding ConcurrentQueue.GetConsumingEnumerable() into that ForEachAsync helper.

Answer 2

Here's the extension method I've created.

    /// <summary>
    /// Concurrently Executes async actions for each item of <see cref="IEnumerable<typeparamref name="T"/>
    /// </summary>
    /// <typeparam name="T">Type of IEnumerable</typeparam>
    /// <param name="enumerable">instance of <see cref="IEnumerable<typeparamref name="T"/>"/></param>
    /// <param name="action">an async <see cref="Action" /> to execute</param>
    /// <param name="maxActionsToRunInParallel">Optional, max numbers of the actions to run in parallel,
    /// Must be grater than 0</param>
    /// <returns>A Task representing an async operation</returns>
    /// <exception cref="ArgumentOutOfRangeException">If the maxActionsToRunInParallel is less than 1</exception>
    public static async Task ForEachAsyncConcurrent<T>(
        this IEnumerable<T> enumerable,
        Func<T, Task> action,
        int? maxActionsToRunInParallel = null)
    {
        if (maxActionsToRunInParallel.HasValue)
        {
            using (var semaphoreSlim = new SemaphoreSlim(
                maxActionsToRunInParallel.Value, maxActionsToRunInParallel.Value))
            {
                var tasksWithThrottler = new List<Task>();

                foreach (var item in enumerable)
                {
                    // Increment the number of currently running tasks and wait if they are more than limit.
                    await semaphoreSlim.WaitAsync();

                    tasksWithThrottler.Add(Task.Run(async () =>
                    {                            
                        await action(item).ContinueWith(res =>
                        {
                            // action is completed, so decrement the number of currently running tasks
                            semaphoreSlim.Release();
                        });
                    }));
                }

                // Wait for all tasks to complete.
                await Task.WhenAll(tasksWithThrottler.ToArray());
            }
        }
        else
        {
            await Task.WhenAll(enumerable.Select(item => action(item)));
        }
    }

Sample Usage:

await enumerable.ForEachAsyncConcurrent(
    async item =>
    {
        await SomeAsyncMethod(item);
    },
    5);