Creating a blocking Queue T in NET

Question

I have a scenario where I have multiple threads adding to a queue and multiple threads reading from the same queue  If the queue reaches a specific size all threads that are filling the queue will be blocked on add until an item is removed from the queue   The solution below is what I am using right now and my question is  How can this be improved   Is there an object that already enables this behavior in the BCL that I should be using   internal class BlockingCollection lt T gt    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  lt  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

User · Answer

I just knocked this up using the Reactive Extensions and remembered this 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;
    }
}

Not necessarily entirely safe, but very simple.

User · Answer

That looks very unsafe  very little synchronization   how about something like   class SizeQueue lt T gt        private readonly Queue lt T gt  queue   new Queue lt T gt         private readonly int maxSize      public SizeQueue int maxSize    this maxSize   maxSize         public void Enqueue T item                lock  queue                        while  queue Count  gt   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                       edit   In reality  you d want a way to close the queue so that readers start exiting cleanly - perhaps something like a bool flag - if set  an empty queue just returns  rather than blocking    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

User · Answer

If you want maximum throughput  allowing multiple readers to read and only one writer to write  BCL has something called ReaderWriterLockSlim that should help slim down your code

User · Answer

This is what I came op for a thread safe bounded blocking queue     using System  using System Collections Generic  using System Text  using System Threading   public class BlockingBuffer lt T gt        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  lt   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

User · Answer

How can this be improved    Well  you need to look at every method in your class and consider what would happen if another thread was simultaneously calling that method or any other method  For example  you put a lock in the Remove method  but not in the Add method  What happens if one thread Adds at the same time as another thread Removes  Bad things   Also consider that a method can return a second object that provides access to the first object s internal data - for example  GetEnumerator  Imagine one thread is going through that enumerator  another thread is modifying the list at the same time  Not good   A good rule of thumb is to make this simpler to get right by cutting down the number of methods in the class to the absolute minimum   In particular  don t inherit another container class  because you will expose all of that class s methods  providing a way for the caller to corrupt the internal data  or to see partially complete changes to the data  just as bad  because the data appears corrupted at that moment   Hide all the details and be completely ruthless about how you allow access to them   I d strongly advise you to use off-the-shelf solutions - get a book about threading or use 3rd party library  Otherwise  given what you re attempting  you re going to be debugging your code for a long time   Also  wouldn t it make more sense for Remove to return an item  say  the one that was added first  as it s a queue   rather than the caller choosing a specific item  And when the queue is empty  perhaps Remove should also block   Update  Marc s answer actually implements all these suggestions     But I ll leave this here as it may be helpful to understand why his version is such an improvement

User · Answer

Well  you might look at System Threading Semaphore class  Other than that - no  you have to make this yourself  AFAIK there is no such built-in collection

User · Answer

Starting with  NET 5 0 Core 3 0 you can use System Threading Channels Benchmarks from this  Asynchronous Producer Consumer Pattern in  NET  C    article show a significant speed boost over BlockingCollection

User · Answer

I haven t fully explored the TPL but they might have something that fits your needs  or at the very least  some Reflector fodder to snag some inspiration from   Hope that helps

User · Answer

Use  net 4 BlockingCollection  to enqueue use Add    to dequeue use Take    It internally uses non-blocking ConcurrentQueue  More info here Fast and Best Producer consumer queue technique BlockingCollection vs concurrent Queue

User · Answer

You can use the BlockingCollection and ConcurrentQueue in the  System Collections Concurrent Namespace   public class ProducerConsumerQueue lt T gt    BlockingCollection lt T gt             lt summary gt          Initializes a new instance of the ProducerConsumerQueue  Use Add and TryAdd for Enqueue and TryEnqueue and Take and TryTake for Dequeue and TryDequeue functionality          lt  summary gt      public ProducerConsumerQueue               base new ConcurrentQueue lt T gt                         lt summary gt        Initializes a new instance of the ProducerConsumerQueue  Use Add and TryAdd for Enqueue and TryEnqueue and Take and TryTake for Dequeue and TryDequeue functionality        lt  summary gt         lt param name  maxSize  gt  lt  param gt      public ProducerConsumerQueue int maxSize            base new ConcurrentQueue lt T gt     maxSize

[c#] Creating a blocking Queue<T> in .NET?

Examples related to c#

Examples related to .net

Examples related to multithreading

Examples related to collections

Examples related to queue