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)
{
}

// The main loop where we generated the data for the call.
{
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());
}

while (true)
{
Guid id;
if (queue.TryDequeue(out id))
{
{
ID = id,
}, ct);
}
else
{
await mutex.DrainQueue(ct);
break;
}
}
}

// Class I'm looking for peer review on
{
private SemaphoreSlim _sem;

: this(10)
{
}

{
}

{
}

{
await _sem.WaitAsync(ct);
try
{
}
catch (OperationCanceledException)
{
// Intentional ignore
return;
}
}

private void _OnCompleted()
{
_sem.Release(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)
{
from partition in Partitioner.Create(source).GetPartitions(dop)
using (partition)
while (partition.MoveNext())
await body(partition.Current);
}));
}


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.

• I'm going to have to give this a look. The ability to cancel a thread or the entire process is a must, but I suspect there is a way to factor that into his implementation for reuse as well. Usually we're talking 10-12 threads with a queue of 500k tasks. The example is simplistic but there is a queue manager that continuously adds to it. – Gary Smith Jan 17 '18 at 6:31
• Yes, cancellation is easy and should be done within ProcessItem like you normally would do it. The loop inherently has no ability to cancel a running work item. Of course you can also test the CancellationToken inside the loop if that would work for you. – usr Jan 23 '18 at 12:06

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>
this IEnumerable<T> enumerable,
int? maxActionsToRunInParallel = null)
{
if (maxActionsToRunInParallel.HasValue)
{
using (var semaphoreSlim = new SemaphoreSlim(
maxActionsToRunInParallel.Value, maxActionsToRunInParallel.Value))
{

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

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

// Wait for all tasks to complete.
}
}
else
{
}
}


Sample Usage:

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

• @Jamal If you read the answer, you'll see that code is well documented. If you still have any queries, feel free to ask. Also I don't think answer is always supposed to be better - I'm posting a different approach - I've used SemaphorSlim instead of Partitioner – Jay Shah May 10 '18 at 0:17
• My mistake, I didn't see it at first. It's not very common here to see entire reviews posted as comments. – Jamal May 10 '18 at 1:01
• Jay, that's an interesting take on it. I have been using the first approach with a few tweaks. In both cases I added some additional logic for passing in a CancellationToken (so Action<T, CancellationToken) and main method CancellationToken ct = default(CancellationToken) (then passing on the inside to the Action<T, CancellationToken>. The cancellation token is a little more critical for me as we might be processing 100k items and we need a mechanism for cleaning stopping them. – Gary Smith May 11 '18 at 23:14
• Yes, you can easily add the CancellationToken parameter and pass it in semaphoreSlim.WaitAsync() & Task,Run's 2nd parameter - which can be helpful to stop the task. – Jay Shah May 14 '18 at 11:10
• This rocks, I love it! – Nicholas Petersen Sep 25 at 20:31