Как асинхронно задерживать все будущие вызовы при обработке исключения

Question

У меня есть асинхронный метод, который использует ограниченный ресурс. Если ресурс становится недоступным, я хочу отложить будущие вызовы, пока ресурс не станет доступен снова.

В принципе, если при доступе к ограниченному ресурсу возникает исключение, один поток обрабатывает ошибку, задерживая все потоки, которые также попадают в исключение, и другие потоки, которые вызывают метод. После 5 секунд поток будет повторять доступ к ресурсу. Это похоже на дросселирование.

Я внедрил это злоупотреблениеTaskCompletionSource в сочетании с SemaphoreSlim. Кажется, что он работает. Может ли это быть улучшено, чтобы быть меньше ... храбрым?

// Use SemaphoreSlim to make sure only one thread handles an error at a time. 
private static readonly SemaphoreSlim mySemaphore = new SemaphoreSlim(1); 

// Use TaskCompletionSource as a flag to delay threads while an error is handled. 
private static volatile TaskCompletionSource<bool> myFlag; 

static MyClass() 
{ 
    myFlag = new TaskCompletionSource<bool>(); 
    myFlag.SetResult(false); // At startup there is no error being handled. 
} 

public async Task DoSomethingAsync() 
{ 
    while (true) 
    { 
     await myFlag.Task; // Wait if an error is being handled. 

     try 
     { 
      await ... // Call some asynchronous operations here that can cause errors. 
      return; 
     } 
     catch 
     { 
      await mySemaphore.WaitAsync(); // Wait so only one thread handles an error. 
      bool wasHandled = await myFlag.Task; // Wait and check if error was handled. 

      if (wasHandled == false) 
      { 
       // Reset TaskCompletionSource so error handling on other threads waits. 
       myFlag = new TaskCompletionSource<bool>(); 
       mySemaphore.Release(); 

       await Task.Delay(5000); // "Handle" the error by waiting 5 seconds. 
       myFlag.SetResult(true); // Notify waiting threads an error was handled. 

       // Reset TaskCompletionSource 
       myFlag = new TaskCompletionSource<bool>(); 
       myFlag.SetResult(false); 
      } 
      else // (wasHandled == true) 
      { 
       mySemaphore.Release(); // Move along, nothing to see here. 
      } 
     } 
    } 
} 

Чтобы пояснить, почему я считаю, что это должно быть улучшено: Я использую TaskCompletionSource создать awaitable логического состояния, и сбросить его, я должен создавать новую TaskCompletionSource каждый раз. Я не думаю, что это предназначено для TaskCompletionSource.

Я просмотрел ManualResetEvent и AutoResetEvent, поскольку они, кажется, делают то, что мне нужно, но они не предлагают асинхронную функциональность.

Answer 1

Некоторые идеи для вас здесь, с образцами кода. Однако вам действительно нужно исследовать шаблон Circuit Breaker, чтобы вы могли реализовать его самостоятельно или получить рабочую реализацию.

Как вы уже сейчас, для вас могут работать функции сброса событий, так как AutoResetEvent покрывает корпус для обработки исключения один за другим (так как он позволяет пропустить только один поток), а ManualResetEventSlim - запуск всех потоков в случае нормального выполнения:

var manual = new ManualResetEventSlim(true); 
var auto = new AutoResetEvent(true); 
while (true) 
{ 
    // check for normal work 
    manual.Wait(); 
    try 
    { 
    } 
    catch 
    { 
     auto.Wait(); 
     // only one thread here 
     // stop all the worker threads 
     manual.Reset(); 
     // handling here 

     // start all the workers 
     manual.Set(); 
     // start exception handlers 
     auto.Set(); 
    } 
} 

Обратите внимание, что вы можете моделировать slim version for auto event with SemaforSlim (как вы уже сделали). Вы должны использовать конструктор SemaforSlim(1, 1), точно для тома 1, чтобы продолжить, и изначально задан semafor.

Тем не менее, эта версия не является асинхронной, поэтому у вас есть выбор.

1. Semafor methods could be awaited, и есть AsyncManualResetEvent implementation from @StephenCleary.

2. Кроме того, вы можете использовать некоторую логику повтора с параметрами тайм-аута для ожидающих циклов. В этом случае вы можете либо yield время выполнения с Thread.Yield методом или, если вы хотите установить тайм-аут повтора, просто используйте Delay task из TPL:

   while (!manual.Wait(SMALL_TIMEOUT_FOR_WAITING) 
   { 
       // Thread.Yield(); 
       await Task.Delay(LARGE_RETRY_TIMEOUT); 
   } 
   

Следует также отметить, что это общее практикуйте до introduce some CancellationToken для ваших циклов и ориентированных на задачи методов, чтобы ваш код стал более управляемым в случае остановки всей системы.

Answer 2

Если я правильно понял вашу ситуацию: Потребитель -> Ваш сервис -> Другая услуга/ресурс.

Вы не должны реализовывать стратегию отступления в своей службе, потому что конечный клиент ждет на другом конце строки. Пусть клиент реализует резервное копирование, вашему сервису просто нужно реализовать автоматический выключатель, и даже это необязательно.

Итак ...

На Сервис стороны:

Вы должны иметь частный рабочий метод (например, DoSomething), которые только заботиться о «что-то делать». Он должен быть обернут публичным методом, который реализует Circuit Breaker pattern.

На стороне 'Consumer':

потребителей вашей 'службы' (общественного) метод должен реализовывать Exponential Backoff Strategy.