Créer une file d'attente de blocage <T> dans .NET?

J'ai un scénario où j'ai plusieurs threads s'ajoutant à une file d'attente et plusieurs threads lisant à partir de la même file d'attente. Si la file d'attente atteint une taille spécifique, tous les threads qui remplissent la file d'attente seront bloqués lors de l'ajout jusqu'à ce qu'un élément soit supprimé de la file d'attente.

La solution ci-dessous est celle que j'utilise actuellement et ma question est la suivante: comment cela peut-il être amélioré? Y a-t-il un objet qui active déjà ce comportement dans la BCL que je devrais utiliser?

internal class BlockingCollection<T> : CollectionBase, IEnumerable
{
    //todo: might be worth changing this into a proper QUEUE

    private AutoResetEvent _FullEvent = new AutoResetEvent(false);

    internal T this[int i]
    {
        get { return (T) List[i]; }
    }

    private int _MaxSize;
    internal int MaxSize
    {
        get { return _MaxSize; }
        set
        {
            _MaxSize = value;
            checkSize();
        }
    }

    internal BlockingCollection(int maxSize)
    {
        MaxSize = maxSize;
    }

    internal void Add(T item)
    {
        Trace.WriteLine(string.Format("BlockingCollection add waiting: {0}", Thread.CurrentThread.ManagedThreadId));

        _FullEvent.WaitOne();

        List.Add(item);

        Trace.WriteLine(string.Format("BlockingCollection item added: {0}", Thread.CurrentThread.ManagedThreadId));

        checkSize();
    }

    internal void Remove(T item)
    {
        lock (List)
        {
            List.Remove(item);
        }

        Trace.WriteLine(string.Format("BlockingCollection item removed: {0}", Thread.CurrentThread.ManagedThreadId));
    }

    protected override void OnRemoveComplete(int index, object value)
    {
        checkSize();
        base.OnRemoveComplete(index, value);
    }

    internal new IEnumerator GetEnumerator()
    {
        return List.GetEnumerator();
    }

    private void checkSize()
    {
        if (Count < MaxSize)
        {
            Trace.WriteLine(string.Format("BlockingCollection FullEvent set: {0}", Thread.CurrentThread.ManagedThreadId));
            _FullEvent.Set();
        }
        else
        {
            Trace.WriteLine(string.Format("BlockingCollection FullEvent reset: {0}", Thread.CurrentThread.ManagedThreadId));
            _FullEvent.Reset();
        }
    }
}

Cela semble très dangereux (très peu de synchronisation); que diriez-vous de quelque chose comme:

class SizeQueue<T>
{
    private readonly Queue<T> queue = new Queue<T>();
    private readonly int maxSize;
    public SizeQueue(int maxSize) { this.maxSize = maxSize; }

    public void Enqueue(T item)
    {
        lock (queue)
        {
            while (queue.Count >= maxSize)
            {
                Monitor.Wait(queue);
            }
            queue.Enqueue(item);
            if (queue.Count == 1)
            {
                // wake up any blocked dequeue
                Monitor.PulseAll(queue);
            }
        }
    }
    public T Dequeue()
    {
        lock (queue)
        {
            while (queue.Count == 0)
            {
                Monitor.Wait(queue);
            }
            T item = queue.Dequeue();
            if (queue.Count == maxSize - 1)
            {
                // wake up any blocked enqueue
                Monitor.PulseAll(queue);
            }
            return item;
        }
    }
}

(Éditer)

En réalité, vous voudriez un moyen de fermer la file d'attente pour que les lecteurs commencent à sortir proprement - peut-être quelque chose comme un indicateur booléen - s'il est défini, une file d'attente vide revient simplement (plutôt que de bloquer):

bool closing;
public void Close()
{
    lock(queue)
    {
        closing = true;
        Monitor.PulseAll(queue);
    }
}
public bool TryDequeue(out T value)
{
    lock (queue)
    {
        while (queue.Count == 0)
        {
            if (closing)
            {
                value = default(T);
                return false;
            }
            Monitor.Wait(queue);
        }
        value = queue.Dequeue();
        if (queue.Count == maxSize - 1)
        {
            // wake up any blocked enqueue
            Monitor.PulseAll(queue);
        }
        return true;
    }
}

Utilisez .net 4 BlockingCollection, pour mettre en file d'attente, utilisez Add (), pour retirer la file, utilisez Take (). Il utilise en interne ConcurrentQueue non bloquant. Plus d'informations ici Technique de file d'attente rapide et meilleure producteur / consommateur BlockingCollection vs file d'attente simultanée

"Comment ça pourrait être amélioré?"

Eh bien, vous devez examiner chaque méthode de votre classe et considérer ce qui se passerait si un autre thread appelait simultanément cette méthode ou toute autre méthode. Par exemple, vous placez un verrou dans la méthode Remove, mais pas dans la méthode Add. Que se passe-t-il si un thread s'ajoute en même temps qu'un autre thread est supprimé? Mauvaises choses.

Considérez également qu'une méthode peut renvoyer un deuxième objet qui fournit l'accès aux données internes du premier objet - par exemple, GetEnumerator. Imaginez qu'un thread passe par cet énumérateur, un autre thread modifie la liste en même temps. Pas bon.

Une bonne règle de base est de simplifier la tâche en réduisant au minimum le nombre de méthodes de la classe.

En particulier, n'héritez pas d'une autre classe de conteneur, car vous exposerez toutes les méthodes de cette classe, ce qui permettra à l'appelant de corrompre les données internes ou de voir les modifications partiellement complètes des données (tout aussi mauvaises, car les données semble corrompu à ce moment). Cachez tous les détails et soyez totalement impitoyable sur la façon dont vous autorisez l'accès à eux.

Je vous conseille fortement d'utiliser des solutions prêtes à l'emploi - obtenir un livre sur le filetage ou utiliser une bibliothèque tierce. Sinon, compte tenu de ce que vous essayez, vous allez déboguer votre code pendant une longue période.

De plus, ne serait-il pas plus logique que Remove renvoie un élément (par exemple, celui qui a été ajouté en premier, car il s'agit d'une file d'attente), plutôt que l'appelant choisissant un élément spécifique? Et lorsque la file d'attente est vide, Remove devrait également se bloquer.

Mise à jour: la réponse de Marc met en œuvre toutes ces suggestions! :) Mais je vais laisser ceci ici car il peut être utile de comprendre pourquoi sa version est une telle amélioration.

Vous pouvez utiliser BlockingCollection et ConcurrentQueue dans l'espace de noms System.Collections.Concurrent

 public class ProducerConsumerQueue<T> : BlockingCollection<T>
{
    /// <summary>
    /// Initializes a new instance of the ProducerConsumerQueue, Use Add and TryAdd for Enqueue and TryEnqueue and Take and TryTake for Dequeue and TryDequeue functionality
    /// </summary>
    public ProducerConsumerQueue()  
        : base(new ConcurrentQueue<T>())
    {
    }

  /// <summary>
  /// Initializes a new instance of the ProducerConsumerQueue, Use Add and TryAdd for Enqueue and TryEnqueue and Take and TryTake for Dequeue and TryDequeue functionality
  /// </summary>
  /// <param name="maxSize"></param>
    public ProducerConsumerQueue(int maxSize)
        : base(new ConcurrentQueue<T>(), maxSize)
    {
    }



}

Je viens de faire tomber ceci en utilisant les extensions réactives et je me suis souvenu de cette question:

public class BlockingQueue<T>
{
    private readonly Subject<T> _queue;
    private readonly IEnumerator<T> _enumerator;
    private readonly object _sync = new object();

    public BlockingQueue()
    {
        _queue = new Subject<T>();
        _enumerator = _queue.GetEnumerator();
    }

    public void Enqueue(T item)
    {
        lock (_sync)
        {
            _queue.OnNext(item);
        }
    }

    public T Dequeue()
    {
        _enumerator.MoveNext();
        return _enumerator.Current;
    }
}

Pas nécessairement entièrement sûr, mais très simple.

C'est ce que je suis venu opter pour une file d'attente de blocage bornée thread safe.

using System;
using System.Collections.Generic;
using System.Text;
using System.Threading;

public class BlockingBuffer<T>
{
    private Object t_lock;
    private Semaphore sema_NotEmpty;
    private Semaphore sema_NotFull;
    private T[] buf;

    private int getFromIndex;
    private int putToIndex;
    private int size;
    private int numItems;

    public BlockingBuffer(int Capacity)
    {
        if (Capacity <= 0)
            throw new ArgumentOutOfRangeException("Capacity must be larger than 0");

        t_lock = new Object();
        buf = new T[Capacity];
        sema_NotEmpty = new Semaphore(0, Capacity);
        sema_NotFull = new Semaphore(Capacity, Capacity);
        getFromIndex = 0;
        putToIndex = 0;
        size = Capacity;
        numItems = 0;
    }

    public void put(T item)
    {
        sema_NotFull.WaitOne();
        lock (t_lock)
        {
            while (numItems == size)
            {
                Monitor.Pulse(t_lock);
                Monitor.Wait(t_lock);
            }

            buf[putToIndex++] = item;

            if (putToIndex == size)
                putToIndex = 0;

            numItems++;

            Monitor.Pulse(t_lock);

        }
        sema_NotEmpty.Release();


    }

    public T take()
    {
        T item;

        sema_NotEmpty.WaitOne();
        lock (t_lock)
        {

            while (numItems == 0)
            {
                Monitor.Pulse(t_lock);
                Monitor.Wait(t_lock);
            }

            item = buf[getFromIndex++];

            if (getFromIndex == size)
                getFromIndex = 0;

            numItems--;

            Monitor.Pulse(t_lock);

        }
        sema_NotFull.Release();

        return item;
    }
}

Je n'ai pas complètement exploré le TPL, mais ils pourraient avoir quelque chose qui correspond à vos besoins, ou à tout le moins, du fourrage Reflector pour s'inspirer.

J'espère que cela pourra aider.

Eh bien, vous pourriez regarder la System.Threading.Semaphoreclasse. A part ça, non, vous devez le faire vous-même. AFAIK il n'y a pas de telle collection intégrée.

Si vous voulez un débit maximal, permettant à plusieurs lecteurs de lire et à un seul écrivain d'écrire, BCL a quelque chose appelé ReaderWriterLockSlim qui devrait vous aider à réduire votre code ...