attente Tâche basée sur la file d'attente

Question

Je me demande s'il existe une implémentation/wrapper pour ConcurrentQueue, similaire à BlockingCollection où prélèvement de la collection ne bloque pas, mais est plutôt asynchrone et provoquera une attente asynchrone jusqu'à ce qu'un élément est placé dans le queue.

J'ai trouvé ma propre implémentation, mais elle ne semble pas fonctionner comme prévu. Je me demande si je réinvente quelque chose qui existe déjà.

Voici mon implémentation:

public class MessageQueue<T> 
{ 
    ConcurrentQueue<T> queue = new ConcurrentQueue<T>(); 

    ConcurrentQueue<TaskCompletionSource<T>> waitingQueue = 
     new ConcurrentQueue<TaskCompletionSource<T>>(); 

    object queueSyncLock = new object(); 

    public void Enqueue(T item) 
    { 
     queue.Enqueue(item); 
     ProcessQueues(); 
    } 

    public async Task<T> Dequeue() 
    { 
     TaskCompletionSource<T> tcs = new TaskCompletionSource<T>(); 
     waitingQueue.Enqueue(tcs); 
     ProcessQueues(); 
     return tcs.Task.IsCompleted ? tcs.Task.Result : await tcs.Task; 
    } 

    private void ProcessQueues() 
    { 
     TaskCompletionSource<T> tcs=null; 
     T firstItem=default(T); 
     while (true) 
     { 
      bool ok; 
      lock (queueSyncLock) 
      { 
       ok = waitingQueue.TryPeek(out tcs) && queue.TryPeek(out firstItem); 
       if (ok) 
       { 
        waitingQueue.TryDequeue(out tcs); 
        queue.TryDequeue(out firstItem); 
       } 
      } 
      if (!ok) break; 
      tcs.SetResult(firstItem); 
     } 
    } 
} 

Answer 1

Je ne sais pas d'une solution sans verrou, mais vous pouvez jeter un oeil à la nouvelle Dataflow library, une partie de la Async CTP. Un simple BufferBlock<T> devrait suffire, .: par exemple

BufferBlock<int> buffer = new BufferBlock<int>(); 

Production et consommation sont les plus faciles à faire par des méthodes d'extension sur les types de blocs de flux de données.

La production est aussi simple que:

buffer.Post(13); 

et la consommation est async prêt:

int item = await buffer.ReceiveAsync(); 

Je ne vous recommande d'utiliser Dataflow si possible; rendre un tel tampon à la fois efficace et correct est plus difficile qu'il n'y paraît d'abord.

Answer 2

Il peut être surpuissant pour votre cas d'utilisation (compte tenu de la courbe d'apprentissage), mais Reactive Extentions fournit toute la colle que vous pourriez jamais vouloir pour la composition asynchrone.

Vous êtes essentiellement abonné aux modifications et elles vous sont transmises dès qu'elles sont disponibles, et vous pouvez demander au système d'insérer les modifications dans un fil distinct.

Answer 3

Vous pouvez simplement utiliser un BlockingCollection (en utilisant la valeur par défaut ConcurrentQueue) et envelopper l'appel à Take dans un Task afin que vous puissiez await il:

var bc = new BlockingCollection<T>(); 

T element = await Task.Run(() => bc.Take()); 

Answer 4

est ici la mise en œuvre que je utilise actuellement.

public class MessageQueue<T> 
{ 
    ConcurrentQueue<T> queue = new ConcurrentQueue<T>(); 
    ConcurrentQueue<TaskCompletionSource<T>> waitingQueue = 
     new ConcurrentQueue<TaskCompletionSource<T>>(); 
    object queueSyncLock = new object(); 
    public void Enqueue(T item) 
    { 
     queue.Enqueue(item); 
     ProcessQueues(); 
    } 

    public async Task<T> DequeueAsync(CancellationToken ct) 
    { 
     TaskCompletionSource<T> tcs = new TaskCompletionSource<T>(); 
     ct.Register(() => 
     { 
      lock (queueSyncLock) 
      { 
       tcs.TrySetCanceled(); 
      } 
     }); 
     waitingQueue.Enqueue(tcs); 
     ProcessQueues(); 
     return tcs.Task.IsCompleted ? tcs.Task.Result : await tcs.Task; 
    } 

    private void ProcessQueues() 
    { 
     TaskCompletionSource<T> tcs = null; 
     T firstItem = default(T); 
     lock (queueSyncLock) 
     { 
      while (true) 
      { 
       if (waitingQueue.TryPeek(out tcs) && queue.TryPeek(out firstItem)) 
       { 
        waitingQueue.TryDequeue(out tcs); 
        if (tcs.Task.IsCanceled) 
        { 
         continue; 
        } 
        queue.TryDequeue(out firstItem); 
       } 
       else 
       { 
        break; 
       } 
       tcs.SetResult(firstItem); 
      } 
     } 
    } 
} 

Il fonctionne assez bien, mais il y a beaucoup de discorde sur queueSyncLock, comme je fais pas mal d'utilisation du CancellationToken d'annuler certaines des tâches d'attente. Bien sûr, cela conduit à beaucoup moins que je verrais le blocage avec un BlockingCollection mais ...

Je me demande s'il y a une plus lisse, verrouiller des moyens libres d'obtenir le même résultat

Answer 5

Mon atempt (il un événement déclenché lorsqu'une « promesse » est créé, et il peut être utilisé par un producteur externe pour savoir quand pour produire plus d'articles):

public class AsyncQueue<T> 
{ 
    private ConcurrentQueue<T> _bufferQueue; 
    private ConcurrentQueue<TaskCompletionSource<T>> _promisesQueue; 
    private object _syncRoot = new object(); 

    public AsyncQueue() 
    { 
     _bufferQueue = new ConcurrentQueue<T>(); 
     _promisesQueue = new ConcurrentQueue<TaskCompletionSource<T>>(); 
    } 

    /// <summary> 
    /// Enqueues the specified item. 
    /// </summary> 
    /// <param name="item">The item.</param> 
    public void Enqueue(T item) 
    { 
     TaskCompletionSource<T> promise; 
     do 
     { 
      if (_promisesQueue.TryDequeue(out promise) && 
       !promise.Task.IsCanceled && 
       promise.TrySetResult(item)) 
      { 
       return;          
      } 
     } 
     while (promise != null); 

     lock (_syncRoot) 
     { 
      if (_promisesQueue.TryDequeue(out promise) && 
       !promise.Task.IsCanceled && 
       promise.TrySetResult(item)) 
      { 
       return; 
      } 

      _bufferQueue.Enqueue(item); 
     }    
    } 

    /// <summary> 
    /// Dequeues the asynchronous. 
    /// </summary> 
    /// <param name="cancellationToken">The cancellation token.</param> 
    /// <returns></returns> 
    public Task<T> DequeueAsync(CancellationToken cancellationToken) 
    { 
     T item; 

     if (!_bufferQueue.TryDequeue(out item)) 
     { 
      lock (_syncRoot) 
      { 
       if (!_bufferQueue.TryDequeue(out item)) 
       { 
        var promise = new TaskCompletionSource<T>(); 
        cancellationToken.Register(() => promise.TrySetCanceled()); 

        _promisesQueue.Enqueue(promise); 
        this.PromiseAdded.RaiseEvent(this, EventArgs.Empty); 

        return promise.Task; 
       } 
      } 
     } 

     return Task.FromResult(item); 
    } 

    /// <summary> 
    /// Gets a value indicating whether this instance has promises. 
    /// </summary> 
    /// <value> 
    /// <c>true</c> if this instance has promises; otherwise, <c>false</c>. 
    /// </value> 
    public bool HasPromises 
    { 
     get { return _promisesQueue.Where(p => !p.Task.IsCanceled).Count() > 0; } 
    } 

    /// <summary> 
    /// Occurs when a new promise 
    /// is generated by the queue 
    /// </summary> 
    public event EventHandler PromiseAdded; 
}