Multiple locking scheme from one manufacturer

Question

I have several producing threads and single consumption. In C #, I use ConcurrentQueue for this.

How can I correctly put the consumer thread to sleep when the queue is empty?

 ManualResetEventSlim signal; void WorkerThread(CancellationToken token) { while(!token.IsCancellationRequested) { object work; if (!_eventQueue.TryDequeue(out work)) { signal.Reset(); signal.Wait(token); continue; } ... } } ... void Produce(object o) { _eventQueue.Enqueue(o); signal.Set(); } 

I tried this, but there is a chance that

• stream B cannot be read from _eventQueue
• stream A is written to _eventQueue
• stream A sets the signal
• stream B resets the signal
• stream B waits indefinitely

How to overcome this? I used to use lock() and Monitor.Wait() . AutoResetEvent may help (it resets a successful Wait ), but it does not support a CancellationToken .

Answer 1

You can use the BlockingCollection class to support multiple producers and one consumer.

Create an object of type BlockingCollection as follows:

 BlockingCollection<object> collection = new BlockingCollection<object>(); //You can have a specific type instead of object if you want 

Manufacturers can simply call the Add method to add an item to the collection as follows:

 collection.Add("value"); 

And the consumer can use the GetConsumingEnumerable method to get an IEnumerable<T> that takes elements from the collection. Such an enumerated block will block (wait for more items) when there are no more items. This method also supports cancellation .

 foreach (var item in collection.GetConsumingEnumerable()) { //Consume item } 

If you call CompleteAdding , then the enumerated consumption will be completed if there are no more elements.

This class is completely thread safe.