Разные HTTP-вызовы, ждут одну и ту же задачу

Question

У меня есть Task, который запускает процесс win, который генерирует файл, если он еще не создан, и возвращает его. Проблема в том, что действие вызывается более одного раза. Точнее, его src атрибут элемента <track>. У меня есть ConcurrentDictionary<Guid, Task<string>>, который отслеживает, для которого Id процесс в настоящее время выполняется

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;
}

В методе действия контроллера

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() метод только начинается Process и ждет его.

process.WaitForExit();

Чего я хочу добиться, так это вернуть результат той же задачи для того же Id. Кажется, что если Id уже существует в словаре, а я await, он запускает другой процесс (снова вызывает метод Run).

Есть ли способ достичь этого?

Answer 1

Как уже указывал уже Жуан Рейс, простого использования метода GetOrAdd недостаточно, чтобы гарантировать, что Task будет создаваться только один раз для каждого ключа. Из документации :

Если вы вызываете GetOrAdd одновременно в разных потоках, valueFactory может быть вызван несколько раз, но только одна пара ключ / значение будет добавлена ​​к словарь.

Быстрый и ленивый способ справиться с этой проблемой - использовать класс Lazy. Вместо хранения Task объектов в словаре вы можете хранить Lazy<Task> оболочки. Таким образом, даже если обертка создается несколько раз для каждого ключа, все посторонние обертки будут отбрасываться без запрашиваемого свойства Value, и, следовательно, без создания повторяющихся задач.

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 2

Вы можете сделать метод atomi c для защиты "опасной зоны":

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();
   }
}

Недостаток здесь в том, что вызовы с различными идентификаторами должны ждать. Так что это делает узким местом. Конечно, вы могли бы сделать усилие, но эй: парни из do tnet сделали это за вас:

Предпочтительно , вы можете использовать GetOrAdd , чтобы сделать то же самое только с методами ConcurrentDictionary:

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.
}

Да, это на самом деле"одна строка". Пожалуйста, посмотрите этот ответ: { ссылка }, из которого я взял исправление для моего некорректного ответа.

Answer 3

Чтобы иметь несколько одновременных вызовов этого метода, но только один для каждого идентификатора, вам нужно использовать ConcurrentDictionary.GetOrAdd с SemaphoreSlim.

GetOrAdd недостаточно, поскольку заводской параметр может быть выполнен более одного раза, см. «Замечания» здесь https://docs.microsoft.com/en-us/dotnet/api/system.collections.concurrent.concurrentdictionary-2.getoradd?view=netframework-4.8

Вот пример:

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