csharptest.net

Again building upon the code from the previous post, this iteration provides some a little more usability. By way of providing a timeout you can TryDequeue and it will do the polling loop for you. If your wondering why I chose to use a polling loop rather than an event signal, I’ll clarify that in my next post on this subject. Right now, this was the fastest and most reliable means to the end. A working producer/consumer for a single reader and single writer that does not use locks.

    class LocklessQueue<T>
    {
        class Item
        {
            public Item Next;
            int _pos, _count;
            readonly T[] _values;
            public Item(int capacity)
            {
                _pos = _count = 0;
                _values = new T[capacity];
                Next = null;
            }
            public bool IsValid { get { return _pos < _count; } }
            public bool PutValue(T value)
            {
                if (_count < _values.Length)
                {
                    _values[_count++] = value;
                    return true;
                }
                return false;
            }
            public T TakeValue()
            {
                int ix = _pos++;
                T value = _values[ix];
                _values[ix] = default(T);
                return value;
            }
        }

        readonly int _allocSize;
        Item _first;
        Item _last;
        private int _maxPollingInterval;

        public LocklessQueue(int allocSize)
        {
            _maxPollingInterval = 1024;
            _allocSize = Math.Max(1, allocSize);
            _first = _last = new Item(_allocSize);
        }

        public int MaxPollingInterval
        {
            get { return _maxPollingInterval; }
            set { if (_maxPollingInterval < 0) throw new ArgumentOutOfRangeException(); _maxPollingInterval = value; }
        }

        public void Enqueue(T value)
        {
            if (!_last.PutValue(value))
            {
                Item i = new Item(_allocSize);
                i.PutValue(value);
                _last.Next = i;
                _last = i;
            }
        }

        public bool IsEmpty
        {
            get
            {
                Item test = _first;
                while (!test.IsValid && test.Next != null)
                    test = test.Next;
                return false == test.IsValid;
            }
        }

        public T Dequeue(int timeout)
        {
            T value;
            if (!TryDequeue(timeout, out value))
                throw new TimeoutException();
            return value;
        }

        public T Dequeue()
        {
            T value;
            if(!TryDequeue(out value))
                throw new InvalidOperationException();
            return value;
        }

        public bool TryDequeue(int timeout, out T value)
        {
            int start = 0, elapsed = 0, sleep = 1;
            if (timeout > 0) start = Environment.TickCount;

            while (!TryDequeue(out value))
            {
                if (timeout == 0)
                    return false;
                if (timeout > 0)
                {
                    elapsed = Environment.TickCount - start;
                    if (elapsed > timeout)
                        return false;
                    if (sleep > (timeout - elapsed))
                        sleep = timeout - elapsed;
                }
                Thread.Sleep(sleep);

                if (sleep < _maxPollingInterval)
                    if((sleep = sleep << 1) > _maxPollingInterval)
                        sleep = _maxPollingInterval;
            }
            return true;
        }

        public bool TryDequeue(out T value)
        {
            while (!_first.IsValid && _first.Next != null)
                _first = _first.Next;

            if (!_first.IsValid)
            {
                value = default(T);
                return false;
            }
            value = _first.TakeValue();
            return true;
        }
    }

Next up… Part 4 – why you should not bother with Lockless Queues (or at least this one)

From time to time I see posts about removing the use of lock() from producer/consumer queues where the number of threads operating are limited. This is an exploration into writing such a queue, the techniques used, and how to create the thread interactions without the use of locks.

Step 1 – Building the first thing that works. Let’s start simple and assume we have a single thread reading from the queue and a single thread writing to the queue. Now we just need to ensure that these two threads don’t write to the same variables. This is actually a very straight-forward problem. We will need a small class to keep a forward-only linked list of the items in the queue. Then we will track both the head and tail of the queue items. So here is the first-pass, the simplest thing that works:

    class LocklessQueue<T>
    {
        class Item
        {
            public Item Next;
            bool _valid;
            T _value;
            public Item(bool valid, T value)
            {
                _valid = valid;
                _value = value;
                Next = null;
            }
            public bool IsValid { get { return _valid; } }
            public T TakeValue()
            {
                T value = _value;
                _valid = false;
                _value = default(T);
                return value;
            }
        }

        Item _first;
        Item _last;

        public LocklessQueue()
        {
            _first = _last = new Item(false, default(T));
        }

        public bool IsEmpty
        {
            get
            {
                while (!_first.IsValid && _first.Next != null)
                    _first = _first.Next;
                return false == _first.IsValid;
            }
        }

        public void Enqueue(T value)
        {
            Item i = new Item(true, value);
            _last.Next = i;
            _last = i;
        }

        public T Dequeue()
        {
            while (!_first.IsValid && _first.Next != null)
                _first = _first.Next;

            if (!_first.IsValid)
                throw new InvalidOperationException();//queue is empty

            return _first.TakeValue();
        }
    }

Now the question is how can we improve upon this? Well one problem this suffers from is the creation of the ‘Item’ instance for each object added to the queue. While this might sound like a small problem it can grow quite large depending upon the frequency of inserts. So the first thing we should look at is reducing the heap thrashing by changing Item to be a collection rather than a single instance. This will be done in a part 2 of this post. Part 3 will enhance it’s usability a little, and attempt to address a mild amount of synchronization between the reader/writer so we can wait for data. Lastly, if I get to it, we can explore how this can be expanded to allow n readers and writers, provided we have a fixed value for ‘n’.