Tworzenie blokującej kolejki Queue <T> w .NET?

Mam scenariusz, w którym mam wiele wątków dodających do kolejki i wiele wątków odczytujących z tej samej kolejki. Jeśli kolejka osiągnie określony rozmiar, wszystkie wątki , które wypełniają kolejkę, zostaną zablokowane przy dodawaniu, dopóki element nie zostanie usunięty z kolejki.

Poniższe rozwiązanie jest tym, czego teraz używam, a moje pytanie brzmi: Jak można to poprawić? Czy istnieje obiekt, który już włącza to zachowanie w BCL, którego powinienem używać?

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

Wygląda to bardzo niebezpiecznie (bardzo mała synchronizacja); co powiesz na coś takiego:

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

(edytować)

W rzeczywistości chciałbyś mieć sposób na zamknięcie kolejki, aby czytelnicy zaczęli czysto wychodzić - być może coś w rodzaju flagi bool - jeśli jest ustawiona, pusta kolejka po prostu wraca (zamiast blokować):

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

Użyj .net 4 BlockingCollection, aby umieścić w kolejce użyj Add (), aby usunąć z kolejki użyj Take (). Wewnętrznie używa nieblokującej ConcurrentQueue. Więcej informacji tutaj Szybka i najlepsza technika kolejkowania producenta / konsumenta BlockingCollection vs Concurrent Queue

„Jak można to poprawić?”

Cóż, musisz przyjrzeć się każdej metodzie w swojej klasie i zastanowić się, co by się stało, gdyby inny wątek jednocześnie wywoływał tę metodę lub inną metodę. Na przykład blokada jest umieszczana w metodzie Remove, ale nie w metodzie Add. Co się stanie, jeśli jeden wątek zostanie dodany w tym samym czasie co inny wątek zostanie usunięty? Złe rzeczy.

Weź również pod uwagę, że metoda może zwrócić drugi obiekt, który zapewnia dostęp do wewnętrznych danych pierwszego obiektu - na przykład GetEnumerator. Wyobraź sobie, że jeden wątek przechodzi przez ten moduł wyliczający, a inny wątek modyfikuje listę w tym samym czasie. Niedobrze.

Dobrą praktyczną zasadą jest uproszczenie tego rozwiązania poprzez zmniejszenie liczby metod w klasie do absolutnego minimum.

W szczególności nie dziedzicz innej klasy kontenera, ponieważ ujawnisz wszystkie metody tej klasy, umożliwiając wywołującemu uszkodzenie wewnętrznych danych lub zobaczenie częściowo kompletnych zmian w danych (tak samo źle, ponieważ dane wydaje się w tym momencie uszkodzony). Ukryj wszystkie szczegóły i bądź całkowicie bezlitosny w kwestii tego, jak udzielasz do nich dostępu.

Zdecydowanie radzę korzystać z gotowych rozwiązań - zdobądź książkę o wątkach lub skorzystaj z biblioteki innej firmy. W przeciwnym razie, biorąc pod uwagę to, czego próbujesz, będziesz debugować swój kod przez długi czas.

Poza tym, czy nie miałoby większego sensu, gdyby funkcja Remove zwracała element (powiedzmy, ten, który został dodany jako pierwszy, ponieważ jest to kolejka), zamiast wybierania przez dzwoniącego określonego elementu? A kiedy kolejka jest pusta, być może Usuń również powinno się zablokować.

Aktualizacja: odpowiedź Marca faktycznie uwzględnia wszystkie te sugestie! :) Ale zostawię to tutaj, ponieważ może być pomocne zrozumienie, dlaczego jego wersja jest tak ulepszona.

Możesz użyć BlockingCollection i ConcurrentQueue w przestrzeni nazw 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)
    {
    }



}

Właśnie podrzuciłem to za pomocą Reactive Extensions i przypomniałem sobie to pytanie:

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

Niekoniecznie całkowicie bezpieczne, ale bardzo proste.

To jest to, co przyszedłem do bezpiecznej, ograniczonej kolejki blokowania wątków.

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

Nie do końca poznałem TPL, ale mogą mieć coś, co pasuje do twoich potrzeb lub przynajmniej trochę pożywienia Reflector, z którego można czerpać inspirację.

Mam nadzieję, że to pomoże.

Cóż, możesz spojrzeć na System.Threading.Semaphoreklasę. Poza tym - nie, musisz to zrobić sam. AFAIK nie ma takiej wbudowanej kolekcji.

Jeśli chcesz mieć maksymalną przepustowość, pozwalającą wielu czytelnikom na czytanie i tylko jednemu pisarzowi do pisania, BCL ma coś, co nazywa się ReaderWriterLockSlim, co powinno pomóc odchudzić twój kod ...