4
\$\begingroup\$

I am developing a high-performance(real-time) system, it requires message passing between its component. speed and accuracy is at most priority. I have explored a few options. ConcurrentQueue<T> is super fast BlockingCollection<T> provides a better API. also BlockingCollection<T> waits for the producer to produce the message. the problem is BlockingCollection<T> is not fast enough for the job. also, it uses locks internally which may lead to locking contention. so I tried to develop a queue that provides blocking behavior(waits for the producer to produce message) and does not use lock internally. here what I have come up with

    public class BlockingConcurrentQueue<T> : IDisposable
    {
        private readonly ConcurrentQueue<T> _internalQueue;
        private AutoResetEvent _autoResetEvent;
        private long _consumed;
        private long _isAddingCompleted = 0;
        private long _produced;
        private long _sleeping;

        public BlockingConcurrentQueue()
        {
            _internalQueue = new ConcurrentQueue<T>();
            _produced = 0;
            _consumed = 0;
            _sleeping = 0;
            _autoResetEvent = new AutoResetEvent(false);
        }

        public bool IsAddingCompleted
        {
            get
            {
                return Interlocked.Read(ref _isAddingCompleted) == 1;
            }
        }

        public bool IsCompleted
        {
            get
            {
                if (Interlocked.Read(ref _isAddingCompleted) == 1 && _internalQueue.IsEmpty)
                    return true;
                else
                    return false;
            }
        }

        public void CompleteAdding()
        {
            Interlocked.Exchange(ref _isAddingCompleted, 1);
        }

        public void Dispose()
        {
            _autoResetEvent.Dispose();
        }

        public void Enqueue(T item)
        {
            _internalQueue.Enqueue(item);

            if (Interlocked.Read(ref _isAddingCompleted) == 1)
                throw new InvalidOperationException("Adding Completed.");

            Interlocked.Increment(ref _produced);
            if (Interlocked.Read(ref _sleeping) == 1)
            {
                Interlocked.Exchange(ref _sleeping, 0);
                _autoResetEvent.Set();
            }
        }

        public bool TryDequeue(out T result)
        {
            if (Interlocked.Read(ref _consumed) == Interlocked.Read(ref _produced))
            {
                Interlocked.Exchange(ref _sleeping, 1);
                _autoResetEvent.WaitOne();
            }

            if (_internalQueue.TryDequeue(out result))
            {
                Interlocked.Increment(ref _consumed);
                return true;
            }
            return false;
        }
    }

I seek your expertise to review

  1. Sequence Guarantee
  2. Any Performance improvement
  3. Am I doing anything wrong which may break in race condition
\$\endgroup\$
5
\$\begingroup\$

This question has a number of misconceptions

  1. You are unlikely to develop a real-time system using .NET because of unbounded GC times
  2. ConcurrentQueue<T> is better than BlockingCollection<T> not because it does not use locks. It actually uses more locks, just in a smarter way.
  3. ConcurrentQueue<T> provides worse API exactly because it is faster. Most of the missing APIs are missing because there is no way of implementing them efficiently and in a thread-safe manner

Your code has a number of weaknesses:

  1. I don't see any reasonable use for IsAddingCompleted and friends

  2. I am not sure there are no race conditions. For example, lets assume we have produced == consumed == 0. Thread1 adds an item and is just before incrementing produced. Thread2 calls TryDequeue, since produced==consumed, it goes through and stops right before changing sleeping. Thread1 continues, changes produced to 1, checks if sleeping is 0, which it is and quits without setting the event. Thread2 continues to Wait(). Now we have a task and an eteranlly-waiting client.

  3. There is no way to abort a waiting TryDequeue and finish gracefully

\$\endgroup\$

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.