¿Crear una cola de bloqueo <T> en .NET?

Tengo un escenario en el que tengo varios subprocesos que se agregan a una cola y múltiples subprocesos que se leen desde la misma cola. Si la cola alcanza un tamaño específico, todos los hilos que llenen la cola se bloquearán al agregarlos hasta que se elimine un elemento de la cola.

La solución a continuación es lo que estoy usando en este momento y mi pregunta es: ¿cómo se puede mejorar esto? ¿Hay algún objeto que ya habilite este comportamiento en el BCL que debería estar usando?

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

Eso parece muy inseguro (muy poca sincronización); ¿Qué tal algo como:

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

(editar)

En realidad, desearía una forma de cerrar la cola para que los lectores comiencen a salir limpiamente, tal vez algo así como una bandera de bool, si se establece, una cola vacía simplemente regresa (en lugar de bloquear):

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

Use .net 4 BlockingCollection, para poner en cola use Add (), para quitar de cola use Take (). Utiliza internamente ConcurrentQueue sin bloqueo. Más información aquí Rápida y mejor técnica de cola de productor / consumidor BlockingCollection vs concurrent Queue

"¿Cómo se puede mejorar esto?"

Bueno, debe mirar cada método en su clase y considerar qué sucedería si otro hilo llamara simultáneamente a ese método o cualquier otro método. Por ejemplo, coloca un bloqueo en el método Eliminar, pero no en el método Agregar. ¿Qué sucede si un hilo agrega al mismo tiempo que otro hilo elimina? Cosas malas.

También considere que un método puede devolver un segundo objeto que proporciona acceso a los datos internos del primer objeto, por ejemplo, GetEnumerator. Imagine que un hilo está pasando por ese enumerador, otro hilo está modificando la lista al mismo tiempo. No está bien.

Una buena regla general es hacer que esto sea más sencillo, reduciendo el número de métodos en la clase al mínimo absoluto.

En particular, no herede otra clase de contenedor, porque expondrá todos los métodos de esa clase, proporcionando una forma para que la persona que llama corrompa los datos internos o vea cambios parcialmente completos en los datos (igual de malo, porque los datos parece corrompido en ese momento). Oculta todos los detalles y sé completamente despiadado sobre cómo permites el acceso a ellos.

Le recomiendo encarecidamente que use soluciones estándar: obtenga un libro sobre subprocesos o use una biblioteca de terceros. De lo contrario, dado lo que está intentando, va a depurar su código durante mucho tiempo.

Además, ¿no tendría más sentido que Eliminar devuelva un elemento (por ejemplo, el que se agregó primero, ya que es una cola), en lugar de que la persona que llama elija un elemento específico? Y cuando la cola está vacía, tal vez Eliminar también debería bloquear.

Actualización: ¡la respuesta de Marc realmente implementa todas estas sugerencias! :) Pero lo dejaré aquí, ya que puede ser útil entender por qué su versión es una mejora.

Puede usar BlockingCollection y ConcurrentQueue en el espacio de nombres 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)
    {
    }



}

Acabo de hacer esto usando las extensiones reactivas y recordé esta pregunta:

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

No necesariamente del todo seguro, pero muy simple.

Esto es lo que encontré para una cola de bloqueo limitada segura para subprocesos.

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

No he explorado completamente el TPL pero pueden tener algo que se adapte a sus necesidades, o al menos, algo de forraje Reflector para inspirarse.

Espero que ayude.

Bueno, podrías mirar la System.Threading.Semaphoreclase. Aparte de eso, no, tienes que hacer esto tú mismo. AFAIK no existe tal colección incorporada.

Si desea un rendimiento máximo, que permita la lectura de varios lectores y la escritura de un solo escritor, BCL tiene algo llamado ReaderWriterLockSlim que debería ayudar a reducir su código ...