Different HTTP calls, await same Task

Question

I have a Task which starts a win process, which generates file if its not created yet and returns it. The problem is that the action is called more than once. To be more precisely its src attribute of a <track> element. I have ConcurrentDictionary<Guid, Task<string>> which keeps track of for which Id a process is currently running

public async Task<string> GenerateVTTFile(Guid Id)
{
  if (_currentGenerators.TryGetValue(id, out Task<string> task))
  {
     return await task; // problem is here?
  }

  var t = Run(); // Task
  _currentGenerators.TryAdd(id, t);

  return await t;
}

In the action method of the controller

var path = await _svc.GetSomePath();
if (string.IsNullOrEmpty(path))
{
    var path = await svc.GenerateVTTFile(id);

    return PhysicalFile(path, "text/vtt");
} 

return PhysicalFile(path, "text/vtt");

Run() method is just starting Process and waits it.

process.WaitForExit();

What I want to achieve is to return the result of the same task for the same Id. It seems that if the Id already exists in the dictionary and I await it starts another process (calls Run method again).

Is there a way to achieve that?

Answer 1

You can make the method atomic to protect the "dangerzone":

private SemaphoreSlim _sem = new SemaphoreSlim(1);

public Task<string> GenerateVTTFile(Guid Id)
{
  _sem.Wait();
  try
  {
     if (_currentGenerators.TryGetValue(Id, out Task<string> task))
     {
        return task;
     }

     var t = Run(); // Task
     _currentGenerators.TryAdd(Id, t); // While Thread 1 is here,
                                       // Thread 2 can already be past the check above ...
                                       // unless we make the method atomic like here.

     return t;
   }
   finally
   {
      _sem.Release();
   }
}

Drawback here is, that also calls with different ids have to wait. So that makes for a bottleneck. Of course, you could make an effort but hey: the dotnet guys did it for you:

Preferably, you can use GetOrAdd to do the same with only ConcurrentDictionary's methods:

public Task<string> GenerateVTTFile(Guid Id)
{
     // EDIT: This overload vv is actually NOT atomic!
     // DO NOT USE: 
     //return _currentGenerators.GetOrAdd(Id, () => Run());
     // Instead:
     return _currentGenerators.GetOrAdd(Id, 
                                        _ => new Lazy<Task<string>>(() => Run(id))).Value;
     // Fix "stolen" from Theodore Zoulias' Answer. Link to his answer below.
     // If you find this helped you, please upvote HIS answer.
}

Yes, it's really a "one-liner". Please see this answer: https://stackoverflow.com/a/61372518/982149 from which I took the fix for my flawed answer.

Answer 2

As pointed out already by João Reis, using simply the GetOrAdd method is not enough to ensure that a Task will be created only once per key. From the documentation:

If you call GetOrAdd simultaneously on different threads, valueFactory may be called multiple times, but only one key/value pair will be added to the dictionary.

The quick and lazy way to deal with this problem is to use the Lazy class. Instead of storing Task objects in the dictionary, you could store Lazy<Task> wrappers. This way even if a wrapper is created multiple times per key, all extraneous wrappers will be discarded without their Value property requested, and so without duplicate tasks created.

private ConcurrentDictionary<Guid, <Lazy<Task<string>>> _currentGenerators;

public Task<string> GenerateVTTFileAsync(Guid id)
{
    return _currentGenerators.GetOrAdd(id,
        _ => new Lazy<Task<string>>(() => Run(id))).Value;
}

Answer 3

In order to have multiple concurrent calls of that method but only one for each id, you need to use ConcurrentDictionary.GetOrAdd with SemaphoreSlim.

GetOrAdd is not enough because the factory parameter might be executed more than once, see "Remarks" here https://learn.microsoft.com/en-us/dotnet/api/system.collections.concurrent.concurrentdictionary-2.getoradd?view=netframework-4.8

Here is an example:

private ConcurrentDictionary<Guid, Generator> _currentGenerators = 
    new ConcurrentDictionary<Guid, Generator>();

public async Task<string> GenerateVTTFile(Guid id)
{
    var generator = _currentGenerators.GetOrAdd(id, _ => new Generator());

    return await generator.RunGenerator().ConfigureAwait(false);
}

public class Generator
{
    private int _started = 0;

    private Task<string> _task;
    private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(1);

    public async Task<string> RunGenerator()
    {
        if (!IsInitialized())
        {
            await Initialize().ConfigureAwait(false);
        }

        return await Interlocked.CompareExchange(ref _task, null, null).ConfigureAwait(false);
    }

    private async Task Initialize()
    {
        await _semaphore.WaitAsync().ConfigureAwait(false);
        try
        {
            // check again after acquiring the lock
            if (IsInitialized())
            {
                return;
            }

            var task = Run();
            _ = Interlocked.Exchange(ref _task, task);
            Interlocked.Exchange(ref _started, 1);
        }
        finally
        {
            _semaphore.Release();
        }
    }

    private bool IsInitialized()
    {
        return Interlocked.CompareExchange(ref _started, 0, 0) == 1;
    }

    private async Task<string> Run()
    {
        // your implementation here
    }
}