170

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<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(); } } }
Improve this question
4
  • 9
    .Net how has built-in classes to help with this scenario. Most of the answers listed here are obsolete. See the most recent answers at the bottom. Look into thread-safe blocking collections. The answers may be obsolete, but it's still a good question! Commented Mar 14, 2013 at 0:29
  • I think it's still a good idea to learn about Monitor.Wait/Pulse/PulseAll even if we have new concurrent classes in .NET. Commented May 14, 2014 at 16:21
  • 1
    Agree with @thewpfguy. You'll want to comprehend the basic locking mechanisms behind the scenes. Also worth noting that Systems.Collections.Concurrent didn't exist until April 2010 and then only in Visual Studio 2010 and above. Definitively not an option for the VS2008 hold outs... Commented Mar 3, 2015 at 20:05
  • 1
    If you're reading this now, take a look at System.Threading.Channels for a multi-writer/multi-reader, bounded, optionally-blocking implementation of this for .NET Core and .NET Standard. Commented Apr 24, 2019 at 11:14

10 Answers 10

Reset to default
206

That looks very unsafe (very little synchronization); how about something like:

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

(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; } }
Improve this answer
Sign up to request clarification or add additional context in comments.

24 Comments

How about changing the wait to a WaitAny and passing in a terminate waithandle on construction ...
@Marc- an optimisation, if you were expecting the queue to always reach capacity, would be to pass the maxSize value into the constructor of the Queue<T>. You could add another constructor to your class to accomodate for that.
Why SizeQueue, why not FixedSizeQueue?
@Lasse - it releases the lock(s) during Wait, so other threads can acquire it. It reclaims the lock(s) when it wakes up.
Nice, as I said, there was something I wasn't getting :) That sure makes me want to revisit some of my thread-code....
|
64

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

Improve this answer

Comments

15

You can use the BlockingCollection and ConcurrentQueue in the System.Collections.Concurrent Namespace

 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) { } }
Improve this answer

3 Comments

BlockingCollection defaults to Queue. So, I don't think this is necessary.
Does BlockingCollection preserve ordering like a queue?
Yes, when it is initialized with a ConcurrentQueue
14

"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.

Improve this answer

Comments

6

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.

Improve this answer

3 Comments

What is Subject<t>? I don't have any resolver for its namespace.
It's part of the Reactive Extensions.
Not an answer. This doesn't answer the question at all.
5

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<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; } }
Improve this answer

2 Comments

Could you provide some code samples of how I'd queue up some thread functions using this library, including how I'd instantiate this class?
This question/response is a bit dated. You should look at the System.Collections.Concurrent namespace for blocking queue support.
2

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.

Improve this answer

1 Comment

I am aware that this is old, but my comment is for newcomers to SO, as OP already knows this today. This is not an answer, this should have been a comment.
2

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!

Improve this answer

Comments

0

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.

Improve this answer

6 Comments

I looked at that for throttling the number of threads that are accessing a resource but it doesn't allow you to block all access to a resource based on some condition (like Collection.Count). AFAIK anyway
Well, you do that part yoursef, just like you do now. Simply instead of MaxSize and _FullEvent you have the Semaphore, which you initialize with the right count in the constructor. Then, upon every Add/Remove you call WaitForOne() or Release().
It's not much different than what you have now. Just more simple IMHO.
Can you give me an example showing this working? I didn't see how to adjust the size of a Semaphor dynamically which this scenario requires. Since you have to be able to block all resources only if the queue is full.
Ahh, changing size! Why didn't you say so immediately? OK, then a semaphore is not for you. Good luck with this approach!
|
-1

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...

Improve this answer

4 Comments

I want none to be able to write if the queue is full though.
So you combine it with a lock. Here are some very good examples albahari.com/threading/part2.aspx#_ProducerConsumerQWaitHandle albahari.com/threading/part4.aspx
With queue/dequeue, everyone is a writer... an exclusive lock will perhaps be more pragmatic
I am aware that this is old, but my comment is for newcomers to SO, as OP already knows this today. This is not an answer, this should have been a comment.

Start asking to get answers

Find the answer to your question by asking.

Ask question

Explore related questions

See similar questions with these tags.