Segnalazioni efficienti Compiti per il completamento di TPL su eventi ricorrenti di frequente

Question

Sto lavorando a un sistema di simulazione che, tra le altre cose, consente l'esecuzione di compiti in fasi temporali simulate. L'esecuzione avviene tutto nel contesto del thread di simulazione, ma, dal punto di vista di un "operatore" che utilizza il sistema, desidera comportarsi in modo asincrono. Per fortuna il TPL, con le comode parole chiave 'async/await', lo rende abbastanza semplice. Ho un metodo primitivo sulla simulazione in questo modo:

public Task CycleExecutedEvent() 
    { 
     lock (_cycleExecutedBroker) 
     { 
      if (!IsRunning) throw new TaskCanceledException("Simulation has been stopped"); 
      return _cycleExecutedBroker.RegisterForCompletion(CycleExecutedEventName); 
     } 
    } 

Questa è fondamentalmente la creazione di una nuova TaskCompletionSource e poi tornare un task. Lo scopo di questa attività è di eseguirne la continuazione quando si verifica il nuovo 'ExecuteCycle' nella simulazione.

Ho poi alcuni metodi di estensione come questo:

public static async Task WaitForDuration(this ISimulation simulation, double duration) 
    { 
     double startTime = simulation.CurrentSimulatedTime; 
     do 
     { 
      await simulation.CycleExecutedEvent(); 
     } while ((simulation.CurrentSimulatedTime - startTime) < duration); 
    } 

    public static async Task WaitForCondition(this ISimulation simulation, Func<bool> condition) 
    { 
     do 
     { 
      await simulation.CycleExecutedEvent(); 
     } while (!condition()); 
    } 

Questi sono molto pratico, quindi, per la costruzione di sequenze da una prospettiva 'operatore', intraprendere azioni basate su condizioni e in attesa per periodi di tempo simulato. Il problema che sto incontrando è che CycleExecuted si verifica molto frequentemente (all'incirca ogni millisecondo se utilizzo a velocità completamente accelerata). Poiché questi metodi di attesa "wait" registrano una nuova "attesa" per ogni ciclo, ciò provoca un notevole turnover nelle istanze di TaskCompletionSource.

Ho profilato il mio codice e ho scoperto che circa il 5,5% del tempo totale della mia CPU viene speso all'interno di questi completamenti, di cui solo una percentuale trascurabile viene spesa nel codice 'attivo'. In pratica, tutto il tempo è dedicato alla registrazione di nuovi completamenti in attesa che le condizioni di attivazione siano valide.

La mia domanda: come posso migliorare le prestazioni qui pur mantenendo la comodità del async/await modello per la scrittura di 'comportamenti' operatore? Sto pensando che ho bisogno di qualcosa come un TaskCompletionSource più leggero e/o riutilizzabile, dato che l'evento di attivazione si verifica così frequentemente.

---

Ho fatto un po 'più di ricerca e suona come una buona opzione potrebbe essere quella di creare un'implementazione personalizzata del modello Awaitable, che potrebbe collegare direttamente l'evento, eliminando la necessità di un po' di TaskCompletionOrganizzazioni di risorse e attività. La ragione per cui potrebbe essere utile qui è che ci sono un sacco di continuazioni diverse in attesa di CycleExecutedEvent e devono essere attese frequentemente. Quindi, idealmente, sto cercando un modo per mettere in coda i callback di continuazione, quindi richiamare tutto in coda ogni volta che si verifica l'evento. Continuerò a scavare, ma accolgo con favore qualsiasi aiuto se la gente conosce un modo pulito per farlo.

---

Per chiunque che stanno guardando questa domanda in futuro, qui è la consuetudine awaiter ho messo insieme:

public sealed class CycleExecutedAwaiter : INotifyCompletion 
{ 
    private readonly List<Action> _continuations = new List<Action>(); 

    public bool IsCompleted 
    { 
     get { return false; } 
    } 

    public void GetResult() 
    { 
    } 

    public void OnCompleted(Action continuation) 
    { 
     _continuations.Add(continuation); 
    } 

    public void RunContinuations() 
    { 
     var continuations = _continuations.ToArray(); 
     _continuations.Clear(); 
     foreach (var continuation in continuations) 
      continuation(); 
    } 

    public CycleExecutedAwaiter GetAwaiter() 
    { 
     return this; 
    } 
} 

E nel simulatore:

private readonly CycleExecutedAwaiter _cycleExecutedAwaiter = new CycleExecutedAwaiter(); 

    public CycleExecutedAwaiter CycleExecutedEvent() 
    { 
     if (!IsRunning) throw new TaskCanceledException("Simulation has been stopped"); 
     return _cycleExecutedAwaiter; 
    } 

E 'un po' buffo, come il cameriere non segnala mai Completato, ma gli incendi continuano a chiamare i completamenti man mano che vengono registrati; ancora, funziona bene per questa applicazione. Ciò riduce il sovraccarico della CPU dal 5,5% al ​​2,1%. Probabilmente richiederà ancora qualche ritocco, ma è un bel miglioramento rispetto all'originale.

Answer 1

La parola chiave await non funziona solo su Task s, funziona su qualsiasi cosa segua lo schema attendibile. Per dettagli, vedere Stephen Toub's article await anything;.

La versione breve è che il tipo deve avere un metodo GetAwaiter() che restituisce un tipo che implementa INotifyCompletion e ha anche IsCompleted proprietà e GetResult() metodo (void -returning, se l'espressione await non dovrebbe avere un valore). Per un esempio, vedere TaskAwaiter.

Se si crea il proprio attendibile, è possibile restituire lo stesso oggetto ogni volta, evitando il sovraccarico di allocare molti TaskCompletionSource s.

Answer 2

avete veramente bisogno di ricevere il WaitForDuration -Event su un thread diverso? In caso contrario, è sufficiente registrare un callback (o un evento) con _cycleExecutedBroker e ricevere la notifica in modo sincrono.Nel callback è possibile testare qualsiasi condizione che si desidera e solo se tale condizione risulta vera, notificare un thread diverso (utilizzando un'attività o un messaggio o qualsiasi altro meccanismo). Capisco che la condizione per cui si esegue il test raramente si valuti, quindi evitate la maggior parte delle chiamate cross-thread in questo modo.

Credo che l'essenza della mia risposta sia: provare a ridurre la quantità di messaggi cross-thread spostando il calcolo sul thread "source".

Answer 3

Ecco la mia versione di ReusableAwaiter simulando TaskCompletionSource

public sealed class ReusableAwaiter<T> : INotifyCompletion 
{ 
    private Action _continuation = null; 
    private T _result = default(T); 
    private Exception _exception = null; 

    public bool IsCompleted 
    { 
     get; 
     private set; 
    } 

    public T GetResult() 
    { 
     if (_exception != null) 
      throw _exception; 
     return _result; 
    } 

    public void OnCompleted(Action continuation) 
    { 
     if (_continuation != null) 
      throw new InvalidOperationException("This ReusableAwaiter instance has already been listened"); 
     _continuation = continuation; 
    } 

    /// <summary> 
    /// Attempts to transition the completion state. 
    /// </summary> 
    /// <param name="result"></param> 
    /// <returns></returns> 
    public bool TrySetResult(T result) 
    { 
     if (!this.IsCompleted) 
     { 
      this.IsCompleted = true; 
      this._result = result; 

      if (_continuation != null) 
       _continuation(); 
      return true; 
     } 
     return false; 
    } 

    /// <summary> 
    /// Attempts to transition the exception state. 
    /// </summary> 
    /// <param name="result"></param> 
    /// <returns></returns> 
    public bool TrySetException(Exception exception) 
    { 
     if (!this.IsCompleted) 
     { 
      this.IsCompleted = true; 
      this._exception = exception; 

      if (_continuation != null) 
       _continuation(); 
      return true; 
     } 
     return false; 
    } 

    /// <summary> 
    /// Reset the awaiter to initial status 
    /// </summary> 
    /// <returns></returns> 
    public ReusableAwaiter<T> Reset() 
    { 
     this._result = default(T); 
     this._continuation = null; 
     this._exception = null; 
     this.IsCompleted = false; 
     return this; 
    } 

    public ReusableAwaiter<T> GetAwaiter() 
    { 
     return this; 
    } 
} 

E qui è il codice di prova.

class Program 
{ 
    static readonly ReusableAwaiter<int> _awaiter = new ReusableAwaiter<int>(); 

    static void Main(string[] args) 
    { 
     Task.Run(() => Test()); 

     Console.ReadLine(); 
     _awaiter.TrySetResult(22); 
     Console.ReadLine(); 
     _awaiter.TrySetException(new Exception("ERR")); 

     Console.ReadLine(); 
    } 

    static async void Test() 
    { 

     int a = await AsyncMethod(); 
     Console.WriteLine(a); 
     try 
     { 
      await AsyncMethod(); 
     } 
     catch(Exception ex) 
     { 
      Console.WriteLine(ex.Message); 
     } 

    } 

    static ReusableAwaiter<int> AsyncMethod() 
    { 
     return _awaiter.Reset(); 
    } 

}