Bounded Queue with multiple consumers and multiple producers It was an interview question:

Should implement a class - BlockingQueue with those constraints:


2 Methods - Adding an item and Take an item from the queue.
The class should support multiple consumers and multiple producers.
The class will getting a bounded size of queue and a maximum of producers and consumers at constructor: 

public BlockingQueue(int boundedCapacity, int producers, int consumers)

At method Take - if the queue is empty - should to wait till have least one item in the queue.
At the method Insert - if the queue is full - should to wait till the queue was dequeued least one item.


I've written a queue class for supporting multiple producers and multiple consumers threads.

the class is getting bounded size and number of producers and consumers.

as following:

class BlockingQueue&lt;T&gt;
{
    readonly ConcurrentQueue&lt;T&gt; _queue = new ConcurrentQueue&lt;T&gt;();
    readonly int _boundedCapacity;

    readonly SemaphoreSlim _semaphoreSlimProducer;
    readonly SemaphoreSlim _semaphoreSlimConsumer;

    public BlockingQueue(int boundedCapacity, int producers, int consumers)
    {
        _boundedCapacity = boundedCapacity;
        _semaphoreSlimProducer = new SemaphoreSlim(0, producers);
        _semaphoreSlimConsumer = new SemaphoreSlim(consumers);
    }

    public void Insert(T item)
    {
        if(item == null) throw new ArgumentNullException("item");
        while(Count &gt;= _boundedCapacity)
        {
            _semaphoreSlimProducer.Wait(1000);
            _semaphoreSlimConsumer.Release();
        }

        _semaphoreSlimProducer.Wait();
        _queue.Enqueue(item);
        _semaphoreSlimConsumer.Release();
    }

    public T Take()
    {
        while(Count == 0)
        {
            _semaphoreSlimConsumer.Wait(1000);
            _semaphoreSlimProducer.Release();
        }
        _semaphoreSlimConsumer.Wait();
        T retval = default(T);

        _queue.TryDequeue(out retval);

        _semaphoreSlimProducer.Release();
        return retval;
    }

    public int Count
    {
        get
        {
            return _queue.Count;
        }
    }
}


Testing:

private static void TestBlockingQueue()
{
    BlockingQueue&lt;string&gt; b = new BlockingQueue&lt;string&gt;(3, 3, 3);

    ConcurrentBag&lt;Task&gt; tasks = new ConcurrentBag&lt;Task&gt;();
    for(int i = 0; i &lt; 10; i++)
    {
        var t = Task.Factory.StartNew(() =&gt; {
            string item = "Threadid:Thread.CurrentThread.ManagedThreadId_" + Thread.CurrentThread.ManagedThreadId + "_Count_";
            int count = 0;
            while(true)
            {
                item += count++;
                b.Insert(item);
                Console.WriteLine("insert item: " + item);
            }
        });
        tasks.Add(t);
    }

    for(int i = 0; i &lt; 10; i++)
    {
        var t = Task.Factory.StartNew(() =&gt; {
            while(true)
            {
                var item = b.Take();
                Console.WriteLine("taked item: " + item);
            }
        });
        tasks.Add(t);
    }

    Task.WaitAll(tasks.ToArray());
}


I will be glad for getting a code review for this implementation.
The implementation clearly has a race condition:


Let's assume Count == _boundedCapacity - 1
Now have 2 producers call Insert simultaneously
Both simultaneously check that the current size is less than the upper bound and move beyond the while loop
Both insert an item.
Now you've exceeded the bounded capacity by 1


While it might be ok to be able to exceed the upper bound by up to the maximum number of producers - 1 the same race occurs in Take. Except on the consumer side it has the nasty side-effect of potentially trying to remove an object from an empty collection. In this case default(T) gets returned which the caller might not expect since Insert doesn't accept null objects.

Based on the null check in Insert I would infer that you never expect this to be instantiated on a value type but only reference types. I don't see any reason for it but if this should really not be allowed then add the type restriction where T: class to the implementation to make it clear.

In the end the bounded capacity really doesn't mean all that much since it's randomly exceeded and one can dequeue randomly from an empty queue.

Overall you should use BlockingCollection&lt;T&gt; to wrap the ConcurrentQueue&lt;T&gt; which implements the bounding capacity correctly. Just make sure to use the Add and Take methods accepting a CancellationToken so blocked operations can get interrupted nicely when you want to shutdown the application.