Limiting the number of concurrent System.Threading.Tasks.Task

Question

I need to perform extensive data inserts to my database. I can implement the code the multithreaded way with a throttled scheduler that limits the number of concurrent operations. On every M rows, a block is formed and inserted into the database as an atomic operation. Multiple concurrent operations shall occur because database is slower than reading and parsing a data file. I often implement this model using multithreading.

If instead I decide to implement my code using await/async (Entity Framework supports asynchronous programming), how can I make sure that no more than N concurrent tasks execute (i.e. go to database) at the same time?

In my initial design, I have instantiated a List<Task>, added new tasks as soon as I read a block of data to be inserted atomically, and then have let my method return after awaiting all of the task. The design-time issue is that the number of concurrent Tasks (and thus memory footprint) are going to explode because tasks are fed faster than they complete for big data files.

I was thinking about using a SemaphoreSlim, but I have little experience with asynchronous programming (unlike multithreaded). So I am asking this question to get feedback about best practices, if there are any.

Answer 1

The design-time issue is that the number of concurrent Tasks (and thus memory footprint) are going to explode because tasks are fed faster than they complete for big data files. I was thinking about using a SemaphoreSlim

Yes, SemaphoreSlim is an appropriate choice for throttling concurrent asynchronous operations:

private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(10);

async Task ThrottledWorkAsync()
{
  await _semaphore.WaitAsync();
  try
  {
    await WorkAsync();
  }
  finally
  {
    _semaphore.Release();
  }
}

However...

If instead I decide to implement my code using await/async (Entity Framework supports asynchronous programming), how can I make sure that no more than N concurrent tasks execute (i.e. go to database) at the same time?

One thing to be aware of is that Entity Framework - while it supports asynchronous APIs - still requires one connection per request. So, you can't have multiple concurrent asynchronous requests with the same DbContext; you'd need to create a separate connection for each concurrent request (or at least N connections that are "borrowed" by the concurrent requests).

Answer 2

If you have at least the n values to insert initially (n being the maximum amount of concurrent Tasks), you can take the following approach:

1. Invoke InsertAsync() n times with the different values.
2. When each of the Task completes, continue with a new call to InsertAsync() (repeat 2).

That way, you won't need to control the concurrency level using a semaphore and will be non-blocking.

I've just published a package that can be useful for this scenarios, it exposes 2 methods Times() and Map(): https://github.com/jorgebay/concurrent-utils

For example:

// Execute MyMethodAsync() 1,000,000 times limiting the maximum amount
// of parallel async operations to 512
await ConcurrentUtils.Times(1000000, 512, (index) => MyMethodAsync(index));

Answer 3

I use this piece of code to execute my threads:

public static async Task WhenAll(this List<Func<Task>> actions, int threadCount)
{
    var executeTaskHelper = new ConcurrentTaskHelper(threadCount);
    return executeTaskHelper.Execute(actions);
}

public class ConcurrentTaskHelper
{
    int _threadCount;
    CountdownEvent _countdownEvent;
    SemaphoreSlim _throttler;

    public ConcurrentTaskHelper(int threadCount)
    {
        _threadCount = threadCount;
         _throttler  = new SemaphoreSlim(threadCount);
    }

    public async Task Execute(List<Func<Task>> tasks)
    {
        _countdownEvent = new CountdownEvent(tasks.Count);

        foreach (var task in tasks)
        {
            await _throttler.WaitAsync();
            Execute(task);
        }

        _countdownEvent.Wait();
    }

    private async Task Execute(Func<Task> task)
    {
        try { await task(); }
        finally { Completed(); }
    }

    private void Completed()
    {
        _throttler.Release();
        _countdownEvent.Signal();
    }
}

This code is based on the code provided in this thread: How to limit the amount of concurrent async I/O operations?

In stead of using the CountdownEvent, it is better to implement the AsyncCountdownEvent. This way it is possible to use _await countdownEvent.WaitAsync();

Calling it should look something like this. It will execute all the tasks, but only 40 (in this case) concurrent:

var tasks = new List<Func<Task>>();
tasks.Add(() => saveAsync());
//add more
await tasks.WhenAll(40);