5
\$\begingroup\$

It was an interview question:

Should implement a class - BlockingQueue with those constraints:

  1. 2 Methods - Adding an item and Take an item from the queue.
  2. The class should support multiple consumers and multiple producers.
  3. 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)
    
  4. At method Take - if the queue is empty - should to wait till have least one item in the queue.

  5. 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<T>
{
    readonly ConcurrentQueue<T> _queue = new ConcurrentQueue<T>();
    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 >= _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<string> b = new BlockingQueue<string>(3, 3, 3);

    ConcurrentBag<Task> tasks = new ConcurrentBag<Task>();
    for(int i = 0; i < 10; i++)
    {
        var t = Task.Factory.StartNew(() => {
            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 < 10; i++)
    {
        var t = Task.Factory.StartNew(() => {
            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.

\$\endgroup\$
7
\$\begingroup\$

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<T> to wrap the ConcurrentQueue<T> 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.

\$\endgroup\$
0
\$\begingroup\$

Thanks for your answer. 1. it's from an interview question that i was asked for - so i cannot use BlockingCollection. 2. the target is implement a structure that can supporting Parallel loop.

i fix the code as follows:

class BlockingQueueV2<T>
    {
        readonly IProducerConsumerCollection<T> _queue = new ConcurrentQueue<T>();
        readonly int _boundedCapacity;

        readonly SemaphoreSlim _semaphoreSlimProducer;
        readonly SemaphoreSlim _semaphoreSlimConsumer;

        readonly int _maxProducers;
        readonly int _maxConsumers;

        public BlockingQueueV2(int boundedCapacity, int producers, int consumers)
        {
            if(boundedCapacity <= 0)
                throw new ArgumentOutOfRangeException("boundedCapacity");

            if(producers <= 0)
                throw new ArgumentOutOfRangeException("producers");

            if(consumers <= 0)
                throw new ArgumentOutOfRangeException("consumers");

            _maxProducers = producers;
            _maxConsumers = consumers;
            _boundedCapacity = boundedCapacity;

            _semaphoreSlimProducer = new SemaphoreSlim(_boundedCapacity, _maxProducers);
            _semaphoreSlimConsumer = new SemaphoreSlim(_boundedCapacity, _maxConsumers);
        }


        public void Insert(T item)
        {

            if(item == null) throw new ArgumentNullException("item");

            if(IsFull)
            {
                try
                {
                    SpinWait wait = new SpinWait();
                    bool isFull = IsFull;
                    while(isFull)
                    {
                        isFull = IsFull;
                        wait.SpinOnce();
                    }
                }
                catch(Exception ex)
                {
                }
            }
            _semaphoreSlimProducer.Wait();
            bool isAdded = false;
            try
            {
                isAdded = _queue.TryAdd(item);
            }
            finally
            {
                if(isAdded && (_semaphoreSlimConsumer.CurrentCount < _maxConsumers && _semaphoreSlimConsumer.CurrentCount >= 0))
                    _semaphoreSlimConsumer.Release();
            }
        }

        public bool Take(out T retval)
        {
            retval = default(T);

            if(IsEmpty)
            {
                try
                {
                    SpinWait wait = new SpinWait();
                    bool isEmpty = IsEmpty;
                    while(isEmpty)
                    {
                        isEmpty = IsEmpty;
                        wait.SpinOnce();
                    }
                }
                catch(Exception ex)
                {
                }
            }

            _semaphoreSlimConsumer.Wait();

            bool isTaked = false;
            try
            {

                isTaked = _queue.TryTake(out retval);
            }
            finally
            {
                if(isTaked && (_semaphoreSlimProducer.CurrentCount < _maxProducers && _semaphoreSlimProducer.CurrentCount >= 0))
                    _semaphoreSlimProducer.Release();
            }
            return isTaked;
        }

        private bool IsFull
        {
            get { return Count > (_boundedCapacity - 1); }
        }

        private bool IsEmpty
        {
            get
            {
                return ((ConcurrentQueue<T>)_queue).IsEmpty;
            }
        }

        public int Count
        {
            get
            {
                if(((ConcurrentQueue<T>)_queue).IsEmpty)
                    return 0;
                return _queue.Count;
            }
        }
    }

BlockingQueueV2<string> b = new BlockingQueueV2<string>(3, 10, 10);
 Parallel.For(0, 10000, i => {
                    try
                    {
                        if((i % 2) == 0)
                        {
                            try
                            {
                                string item = "ManagedThreadId: " + Thread.CurrentThread.ManagedThreadId + " item: " + i;
                                b.Insert(item);
                                Console.WriteLine("insert item: " + item);
                            }
                            catch(Exception)
                            {

                                throw;
                            }
                        }
                        else
                        {
                            try
                            {
                                string item;
                                bool isTaked = b.Take(out item);
                                Console.WriteLine("taked item: " + item + " isTaked: " + isTaked);
                            }
                            catch(Exception)
                            {

                                throw;
                            }
                        }
                    }
                    catch(Exception ex)
                    {
                        Console.WriteLine(ex.Message);
                    }
                });

It seems that for parallel the above code is working perfectly, but with threads it's not working good..

  BlockingQueueV2<string> b = new BlockingQueueV2<string>(3, 10, 10);
                //BlockingCollection<string> b = new BlockingCollection<string>(3);
                ConcurrentBag<Task> tasks = new ConcurrentBag<Task>();

                for(int i = 0; i < 10; i++)
                {
                    var t1 = Task.Factory.StartNew(() => {
                        try
                        {
                            bool isTaked = true;
                            while(isTaked)
                            {
                                string item;
                               // = b.Take(out item);
                                isTaked = b.Take(out item);
                                Console.WriteLine("taked item: " + item + " isTaked: " + isTaked);
                            }
                        }
                        catch(Exception)
                        {

                            throw;
                        }
                    });
                    tasks.Add(t1);
                }


                for(int i = 0; i < 10; i++)
                {
                    var t = Task.Factory.StartNew(() => {
                        try
                        {
                            string item = "ManagedThreadId: " + Thread.CurrentThread.ManagedThreadId + " item: " + i;
                            //b.Insert(item);
                            b.Insert(item);
                            Console.WriteLine("insert item: " + item);
                        }
                        catch(Exception)
                        {
                            throw;
                        }

                    });
                    tasks.Add(t);
                }



                Task.WaitAll(tasks.ToArray());

Update::

I have a few changes:

  • fix semaphors:

    _semaphoreSlimProducer = new SemaphoreSlim(_boundedCapacity, _maxProducers); _semaphoreSlimConsumer = new SemaphoreSlim(0, _maxConsumers);

  • Insert method:

Now, after _semaphoreSlimProducer.Wait(); I adding loop with check the current count (for blocking adding item to the full queue) as follows:

int _currentCount = 0;
public void Insert(T item)
{
    if(item == null) throw new ArgumentNullException("item");
    _semaphoreSlimProducer.Wait();
    SpinWait spin = new SpinWait();
    while(true)
    {
        int oldCount = _currentCount;
        if(oldCount >= _boundedCapacity)
            continue;
        if(Interlocked.CompareExchange(ref _currentCount, oldCount + 1, oldCount) == oldCount)
            break;
        spin.SpinOnce();
    }
    bool isAdding = false;
    try
    {
        isAdding = _queue.TryAdd(item);
    }
    catch(Exception)
    {
        _semaphoreSlimProducer.Release();
    }
    finally
    {
        if(isAdding)
            _semaphoreSlimConsumer.Release();
    }
}
  • Take method:

Now, after _semaphoreSlimConsumer.Wait(); I adding loop with check the current count (for blocking taking item from empty queue) as follows:

public bool Take(out T retval)
{
    retval = default(T);
    _semaphoreSlimConsumer.Wait();
    SpinWait spin = new SpinWait();
    while(true)
    {
        int oldCount = _currentCount;
        if(oldCount == 0)
            continue;
        if(Interlocked.CompareExchange(ref _currentCount, oldCount - 1, oldCount) == oldCount)
            break;
        spin.SpinOnce();
    }
    bool isTaked = false;
    try
    {
        isTaked = _queue.TryTake(out retval);
    }
    catch(Exception)
    {
        _semaphoreSlimConsumer.Release();
    }
    finally
    {
        if(isTaked)
            _semaphoreSlimProducer.Release();
    }
    return isTaked;
}
\$\endgroup\$
  • \$\begingroup\$ We have a special tag for it, the interview-question which you should have used and also telling us about the constraint in your question :-) \$\endgroup\$ – t3chb0t Aug 13 '17 at 16:14
  • 1
    \$\begingroup\$ I see you've changed the while loops so they don't so readily crash, however I still don't see the point of them... they don't seem to achieve anything, and the checks themselves are unreliable as ChrisWue has said. You've added a lot of unhelpful checks at the end of Insert, and it looks like _semaphoreSlimConsumer = new SemaphoreSlim(_boundedCapacity, _maxConsumers); is wrong (should be 0 initially, similar issues with your old code). As far as I can tell, removing the while loops and fixing the constructor would have made your old code fine, may be issues with queue freshness \$\endgroup\$ – VisualMelon Aug 13 '17 at 18:37
  • \$\begingroup\$ @t3chb0t, cool, i didn't know that this tag is exists - so i adding that, i will add the contraints (hope that i will remember all of them). \$\endgroup\$ – David Michaeli Aug 14 '17 at 11:29

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.