ConcurrentQueue.cs source code in C# .NET

Source code for the .NET framework in C#

                        

Code:

/ 4.0 / 4.0 / DEVDIV_TFS / Dev10 / Releases / RTMRel / ndp / clr / src / BCL / System / Collections / Concurrent / ConcurrentQueue.cs / 1305376 / ConcurrentQueue.cs

                            #pragma warning disable 0420 

// ==++==
//
//   Copyright (c) Microsoft Corporation.  All rights reserved. 
//
// ==--== 
// =+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+ 
//
// ConcurrentQueue.cs 
//
// [....]
//
// A lock-free, concurrent queue primitive, and its associated debugger view type. 
//
// =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=- 
 
using System;
using System.Collections; 
using System.Collections.Generic;
using System.Diagnostics;
using System.Diagnostics.Contracts;
using System.Runtime.ConstrainedExecution; 
using System.Runtime.InteropServices;
using System.Runtime.Serialization; 
using System.Security; 
using System.Security.Permissions;
using System.Threading; 

namespace System.Collections.Concurrent
{
 
    /// 
    /// Represents a thread-safe first-in, first-out collection of objects. 
    ///  
    /// Specifies the type of elements in the queue.
    ///  
    /// All public  and protected members of  are thread-safe and may be used
    /// concurrently from multiple threads.
    /// 
    [ComVisible(false)] 
    [DebuggerDisplay("Count = {Count}")]
    [DebuggerTypeProxy(typeof(SystemCollectionsConcurrent_ProducerConsumerCollectionDebugView<>))] 
    [HostProtection(Synchronization = true, ExternalThreading = true)] 
    [Serializable]
    public class ConcurrentQueue : IProducerConsumerCollection 
    {
        //fields of ConcurrentQueue
        [NonSerialized]
        private volatile Segment m_head; 

        [NonSerialized] 
        private volatile Segment m_tail; 

        private T[] m_serializationArray; // Used for custom serialization. 

        private const int SEGMENT_SIZE = 32;

        ///  
        /// Initializes a new instance of the  class.
        ///  
        public ConcurrentQueue() 
        {
            m_head = m_tail = new Segment(0); 
        }

        /// 
        /// Initializes the contents of the queue from an existing collection. 
        /// 
        /// A collection from which to copy elements. 
        private void InitializeFromCollection(IEnumerable collection) 
        {
            m_head = m_tail = new Segment(0); 
            int index = 0;
            foreach (T element in collection)
            {
                Contract.Assert(index >= 0 && index < SEGMENT_SIZE); 
                m_tail.UnsafeAdd(element);
                index++; 
 
                if (index >= SEGMENT_SIZE)
                { 
                    m_tail = m_tail.UnsafeGrow();
                    index = 0;
                }
            } 
        }
 
        ///  
        /// Initializes a new instance of the 
        /// class that contains elements copied from the specified collection 
        /// 
        /// The collection whose elements are copied to the new .
        /// The  argument is 
        /// null.
        public ConcurrentQueue(IEnumerable collection) 
        { 
            if (collection == null)
            { 
                throw new ArgumentNullException("collection");
            }

            InitializeFromCollection(collection); 
        }
 
        ///  
        /// Get the data array to be serialized
        ///  
        [OnSerializing]
        private void OnSerializing(StreamingContext context)
        {
            // save the data into the serialization array to be saved 
            m_serializationArray = ToArray();
        } 
 
        /// 
        /// Construct the queue from a previously seiralized one 
        /// 
        [OnDeserialized]
        private void OnDeserialized(StreamingContext context)
        { 
            Contract.Assert(m_serializationArray != null);
            InitializeFromCollection(m_serializationArray); 
            m_serializationArray = null; 
        }
 
        /// 
        /// Copies the elements of the  to an , starting at a particular
        ///  index. 
        /// 
        /// The one-dimensional Array that is the 
        /// destination of the elements copied from the 
        /// . The Array must have zero-based indexing. 
        /// The zero-based index in  at which copying
        /// begins.
        ///  is a null reference (Nothing in
        /// Visual Basic). 
        ///  is less than
        /// zero. 
        ///  
        ///  is multidimensional. -or-
        ///  does not have zero-based indexing. -or- 
        ///  is equal to or greater than the length of the 
        /// -or- The number of elements in the source  is
        /// greater than the available space from  to the end of the destination
        /// . -or- The type of the source  cannot be cast automatically to the type of the
        /// destination . 
        ///  
        void ICollection.CopyTo(Array array, int index)
        { 
            // Validate arguments.
            if (array == null)
            {
                throw new ArgumentNullException("array"); 
            }
 
            // We must be careful not to corrupt the array, so we will first accumulate an 
            // internal list of elements that we will then copy to the array. This requires
            // some extra allocation, but is necessary since we don't know up front whether 
            // the array is sufficiently large to hold the stack's contents.
            ((ICollection)ToList()).CopyTo(array, index);
        }
 
        /// 
        /// Gets a value indicating whether access to the  is 
        /// synchronized with the SyncRoot. 
        /// 
        /// true if access to the  is synchronized 
        /// with the SyncRoot; otherwise, false. For , this property always
        /// returns false.
        bool ICollection.IsSynchronized
        { 
            // Gets a value indicating whether access to this collection is synchronized. Always returns
            // false. The reason is subtle. While access is in face thread safe, it's not the case that 
            // locking on the SyncRoot would have prevented concurrent pushes and pops, as this property 
            // would typically indicate; that's because we internally use CAS operations vs. true locks.
            get { return false; } 
        }


        ///  
        /// Gets an object that can be used to synchronize access to the . This property is not supported. 
        ///  
        /// The SyncRoot property is not supported.
        object ICollection.SyncRoot 
        {
            get
            {
                throw new NotSupportedException(Environment.GetResourceString("ConcurrentCollection_SyncRoot_NotSupported")); 
            }
        } 
 
        /// 
        /// Returns an enumerator that iterates through a collection. 
        /// 
        /// An  that can be used to iterate through the collection.
        IEnumerator IEnumerable.GetEnumerator()
        { 
            return ((IEnumerable)this).GetEnumerator();
        } 
 
        /// 
        /// Attempts to add an object to the .
        /// 
        /// The object to add to the . The value can be a null 
        /// reference (Nothing in Visual Basic) for reference types.
        ///  
        /// true if the object was added successfully; otherwise, false. 
        /// For , this operation will always add the object to the
        /// end of the  
        /// and return true.
        bool IProducerConsumerCollection.TryAdd(T item)
        {
            Enqueue(item); 
            return true;
        } 
 
        /// 
        /// Attempts to remove and return an object from the .
        /// 
        /// 
        /// When this method returns, if the operation was successful,  contains the 
        /// object removed. If no object was available to be removed, the value is unspecified.
        ///  
        /// true if an element was removed and returned succesfully; otherwise, false. 
        /// For , this operation will attempt to remove the object
        /// from the beginning of the . 
        /// 
        bool IProducerConsumerCollection.TryTake(out T item)
        {
            return TryDequeue(out item); 
        }
 
        ///  
        /// Gets a value that indicates whether the  is empty.
        ///  
        /// true if the  is empty; otherwise, false.
        /// 
        /// For determining whether the collection contains any items, use of this property is recommended
        /// rather than retrieving the number of items from the  property and comparing it 
        /// to 0.  However, as this collection is intended to be accessed concurrently, it may be the case
        /// that another thread will modify the collection after  returns, thus invalidating 
        /// the result. 
        /// 
        public bool IsEmpty 
        {
            get
            {
                Segment head = m_head; 
                if (!head.IsEmpty)
                    //fast route 1: 
                    //if current head is not empty, then queue is not empty 
                    return false;
                else if (head.Next == null) 
                    //fast route 2:
                    //if current head is empty and it's the last segment
                    //then queue is empty
                    return true; 
                else
                //slow route: 
                //current head is empty and it is NOT the last segment, 
                //it means another thread is growing new segment
                { 
                    SpinWait spin = new SpinWait();
                    while (head.IsEmpty)
                    {
                        if (head.Next == null) 
                            return true;
 
                        spin.SpinOnce(); 
                        head = m_head;
                    } 
                    return false;
                }
            }
        } 

        ///  
        /// Copies the elements stored in the  to a new array. 
        /// 
        /// A new array containing a snapshot of elements copied from the .
        public T[] ToArray()
        {
            return ToList().ToArray(); 
        }
 
        ///  
        /// Copies the  elements to a new . 
        /// 
        /// A new  containing a snapshot of
        /// elements copied from the .
        private List ToList() 
        {
            //store head and tail positions in buffer, 
            Segment head, tail; 
            int headLow, tailHigh;
            GetHeadTailPositions(out head, out tail, out headLow, out tailHigh); 

            if (head == tail)
            {
                return head.ToList(headLow, tailHigh); 
            }
 
            List list = new List(head.ToList(headLow, SEGMENT_SIZE - 1)); 
            Segment curr = head.Next;
            while (curr != tail) 
            {
                list.AddRange(curr.ToList(0, SEGMENT_SIZE - 1));
                curr = curr.Next;
            } 
            //Add tail segment
            list.AddRange(tail.ToList(0, tailHigh)); 
            return list; 
        }
 
        /// 
        /// Store the position of the current head and tail positions.
        /// 
        /// return the head segment 
        /// return the tail segment
        /// return the head offset 
        /// return the tail offset 
        private void GetHeadTailPositions(out Segment head, out Segment tail,
            out int headLow, out int tailHigh) 
        {
            head = m_head;
            tail = m_tail;
            headLow = head.Low; 
            tailHigh = tail.High;
            SpinWait spin = new SpinWait(); 
 
            //we loop until the observed values are stable and sensible.
            //This ensures that any update order by other methods can be tolerated. 
            while (
                //if head and tail changed, retry
                head != m_head || tail != m_tail
                //if low and high pointers, retry 
                || headLow != head.Low || tailHigh != tail.High
                //if head jumps ahead of tail because of concurrent grow and dequeue, retry 
                || head.m_index > tail.m_index) 
            {
                spin.SpinOnce(); 
                head = m_head;
                tail = m_tail;
                headLow = head.Low;
                tailHigh = tail.High; 
            }
        } 
 

        ///  
        /// Gets the number of elements contained in the .
        /// 
        /// The number of elements contained in the .
        ///  
        /// For determining whether the collection contains any items, use of the 
        /// property is recommended rather than retrieving the number of items from the  
        /// property and comparing it to 0. 
        /// 
        public int Count 
        {
            get
            {
                //store head and tail positions in buffer, 
                Segment head, tail;
                int headLow, tailHigh; 
                GetHeadTailPositions(out head, out tail, out headLow, out tailHigh); 

                if (head == tail) 
                {
                    return tailHigh - headLow + 1;
                }
 
                //head segment
                int count = SEGMENT_SIZE - headLow; 
 
                //middle segment(s), if any, are full.
                //We don't deal with overflow to be consistent with the behavior of generic types in CLR. 
                count += SEGMENT_SIZE * ((int)(tail.m_index - head.m_index - 1));

                //tail segment
                count += tailHigh + 1; 

                return count; 
            } 
        }
 

        /// 
        /// Copies the  elements to an existing one-dimensional Array, starting at the specified array index. 
        /// 
        /// The one-dimensional Array that is the 
        /// destination of the elements copied from the 
        /// . The Array must have zero-based
        /// indexing. 
        /// The zero-based index in  at which copying
        /// begins.
        ///  is a null reference (Nothing in
        /// Visual Basic). 
        ///  is less than
        /// zero. 
        ///  is equal to or greater than the 
        /// length of the 
        /// -or- The number of elements in the source  is greater than the 
        /// available space from  to the end of the destination .
        /// 
        public void CopyTo(T[] array, int index) 
        {
            if (array == null) 
            { 
                throw new ArgumentNullException("array");
            } 

            // We must be careful not to corrupt the array, so we will first accumulate an
            // internal list of elements that we will then copy to the array. This requires
            // some extra allocation, but is necessary since we don't know up front whether 
            // the array is sufficiently large to hold the stack's contents.
            ToList().CopyTo(array, index); 
        } 

 
        /// 
        /// Returns an enumerator that iterates through the .
        ///  
        /// An enumerator for the contents of the . 
        ///  
        /// The enumeration represents a moment-in-time snapshot of the contents
        /// of the queue.  It does not reflect any updates to the collection after 
        ///  was called.  The enumerator is safe to use
        /// concurrently with reads from and writes to the queue.
        /// 
        public IEnumerator GetEnumerator() 
        {
            return ToList().GetEnumerator(); 
        } 

        ///  
        /// Adds an object to the end of the .
        /// 
        /// The object to add to the end of the . The value can be a null reference 
        /// (Nothing in Visual Basic) for reference types.
        ///  
        public void Enqueue(T item) 
        {
            SpinWait spin = new SpinWait(); 
            while (true)
            {
                Segment tail = m_tail;
                if (tail.TryAppend(item, ref m_tail)) 
                    return;
                spin.SpinOnce(); 
            } 
        }
 

        /// 
        /// Attempts to remove and return the object at the beginning of the . 
        /// 
        ///  
        /// When this method returns, if the operation was successful,  contains the 
        /// object removed. If no object was available to be removed, the value is unspecified.
        ///  
        /// true if an element was removed and returned from the beggining of the 
        /// succesfully; otherwise, false.
        public bool TryDequeue(out T result) 
        {
            while (!IsEmpty) 
            { 
                Segment head = m_head;
                if (head.TryRemove(out result, ref m_head)) 
                    return true;
                //since method IsEmpty spins, we don't need to spin in the while loop
            }
            result = default(T); 
            return false;
        } 
 
        /// 
        /// Attempts to return an object from the beginning of the  
        /// without removing it.
        /// 
        /// When this method returns,  contains an object from
        /// the beginning of the  or an 
        /// unspecified value if the operation failed.
        /// true if and object was returned successfully; otherwise, false. 
        public bool TryPeek(out T result) 
        {
            while (!IsEmpty) 
            {
                Segment head = m_head;
                if (head.TryPeek(out result))
                    return true; 
                //since method IsEmpty spins, we don't need to spin in the while loop
            } 
            result = default(T); 
            return false;
        } 


        /// 
        /// private class for ConcurrentQueue. 
        /// a queue is a linked list of small arrays, each node is called a segment.
        /// A segment contains an array, a pointer to the next segment, and m_low, m_high indices recording 
        /// the first and last valid elements of the array. 
        /// 
        private class Segment 
        {
            //we define two volatile arrays: m_array and m_state. Note that the accesses to the array items
            //do not get volatile treatment. But we don't need to worry about loading adjacent elements or
            //store/load on adjacent elements would suffer reordering. 
            // - Two stores:  these are at risk, but CLRv2 memory model guarantees store-release hence we are safe.
            // - Two loads: because one item from two volatile arrays are accessed, the loads of the array references 
            //          are sufficient to prevent reordering of the loads of the elements. 
            internal volatile T[] m_array;
 
            //Each array entry has 2 possible states
            // 0 -- initial
            // 1 -- contains value: it may be dequeued, but value is still accessible
            // todo: change this int array to bitmap 
            private volatile int[] m_state;
 
            //pointer to the next segment. null if the current segment is the last segment 
            private volatile Segment m_next;
 
            //We use this zero based index to track how many segments have been created for the queue, and
            //to compute how many active segments are there currently.
            // * The number of currently active segments is : m_tail.m_index - m_head.m_index + 1;
            // * m_index is incremented with every Segment.Grow operation. We use Int64 type, and we can safely 
            //   assume that it never overflows. To overflow, we need to do 2^63 increments, even at a rate of 4
            //   billion (2^32) increments per second, it takes 2^31 seconds, which is about 64 years. 
            internal readonly long m_index; 

            //indices of where the first and last valid values 
            // - m_low points to the position of the next element to pop from this segment, range [0, infinity)
            //      m_low >= SEGMENT_SIZE implies the segment is disposable
            // - m_high points to the position of the latest pushed element, range [-1, infinity)
            //      m_high == -1 implies the segment is new and empty 
            //      m_high >= SEGMENT_SIZE-1 means this segment is ready to grow.
            //        and the thread who sets m_high to SEGMENT_SIZE-1 is responsible to grow the segment 
            // - Math.Min(m_low, SEGMENT_SIZE) > Math.Min(m_high, SEGMENT_SIZE-1) implies segment is empty 
            // - initially m_low =0 and m_high=-1;
            private volatile int m_low; 
            private volatile int m_high;

            /// 
            /// Create and initialize a segment with the specified index. 
            /// 
            internal Segment(long index) 
            { 
                m_array = new T[SEGMENT_SIZE];
                m_state = new int[SEGMENT_SIZE]; //all initialized to 0 
                m_high = -1;
                Contract.Assert(index >= 0);
                m_index = index;
            } 

            ///  
            /// return the next segment 
            /// 
            internal Segment Next 
            {
                get { return m_next; }
            }
 

            ///  
            /// return true if the current segment is empty (doesn't have any element available to dequeue, 
            /// false otherwise
            ///  
            internal bool IsEmpty
            {
                get { return (Low > High); }
            } 

            ///  
            /// Add an element to the tail of the current segment 
            /// exclusively called by ConcurrentQueue.InitializedFromCollection
            /// InitializeFromCollection is responsible to guaratee that there is no index overflow, 
            /// and there is no contention
            /// 
            /// 
            internal void UnsafeAdd(T value) 
            {
                Contract.Assert(m_high < SEGMENT_SIZE - 1); 
                m_high++; 
                m_array[m_high] = value;
                m_state[m_high] = 1; 
            }

            /// 
            /// Create a new segment and append to the current one 
            /// Does not update the m_tail pointer
            /// exclusively called by ConcurrentQueue.InitializedFromCollection 
            /// InitializeFromCollection is responsible to guaratee that there is no index overflow, 
            /// and there is no contention
            ///  
            /// the reference to the new Segment
            internal Segment UnsafeGrow()
            {
                Contract.Assert(m_high >= SEGMENT_SIZE - 1); 
                Segment newSegment = new Segment(m_index + 1); //m_index is Int64, we don't need to worry about overflow
                m_next = newSegment; 
                return newSegment; 
            }
 
            /// 
            /// Create a new segment and append to the current one
            /// Update the m_tail pointer
            /// This method is called when there is no contention 
            /// 
            internal void Grow(ref Segment tail) 
            { 
                //no CAS is needed, since there is no contention (other threads are blocked, busy waiting)
                Segment newSegment = new Segment(m_index + 1);  //m_index is Int64, we don't need to worry about overflow 
                m_next = newSegment;
                Contract.Assert(tail == this);
                tail = m_next;
            } 

 
            ///  
            /// Try to append an element at the end of this segment.
            ///  
            /// the element to append
            /// The tail.
            /// true if the element is appended, false if the current segment is full
            /// if appending the specified element succeeds, and after which the segment is full, 
            /// then grow the segment
            internal bool TryAppend(T value, ref Segment tail) 
            { 
                //quickly check if m_high is already over the boundary, if so, bail out
                if (m_high >= SEGMENT_SIZE - 1) 
                {
                    return false;
                }
 
                //Now we will use a CAS to increment m_high, and store the result in newhigh.
                //Depending on how many free spots left in this segment and how many threads are doing this Increment 
                //at this time, the returning "newhigh" can be 
                // 1) < SEGMENT_SIZE - 1 : we took a spot in this segment, and not the last one, just insert the value
                // 2) == SEGMENT_SIZE - 1 : we took the last spot, insert the value AND grow the segment 
                // 3) > SEGMENT_SIZE - 1 : we failed to reserve a spot in this segment, we return false to
                //    Queue.Enqueue method, telling it to try again in the next segment.

                int newhigh = SEGMENT_SIZE; //initial value set to be over the boundary 

                //We need do Interlocked.Increment and value/state update in a finally block to ensure that they run 
                //without interuption. This is to prevent anything from happening between them, and another dequeue 
                //thread maybe spinning forever to wait for m_state[] to be 1;
                try 
                { }
                finally
                {
                    newhigh = Interlocked.Increment(ref m_high); 
                    if (newhigh <= SEGMENT_SIZE - 1)
                    { 
                        m_array[newhigh] = value; 
                        m_state[newhigh] = 1;
                    } 

                    //if this thread takes up the last slot in the segment, then this thread is responsible
                    //to grow a new segment. Calling Grow must be in the finally block too for reliability reason:
                    //if thread abort during Grow, other threads will be left busy spinning forever. 
                    if (newhigh == SEGMENT_SIZE - 1)
                    { 
                        Grow(ref tail); 
                    }
                } 

                //if newhigh <= SEGMENT_SIZE-1, it means the current thread successfully takes up a spot
                return newhigh <= SEGMENT_SIZE - 1;
            } 

 
            ///  
            /// try to remove an element from the head of current segment
            ///  
            /// The result.
            /// The head.
            /// return false only if the current segment is empty
            internal bool TryRemove(out T result, ref Segment head) 
            {
                SpinWait spin = new SpinWait(); 
                int lowLocal = Low, highLocal = High; 
                while (lowLocal <= highLocal)
                { 
                    //try to update m_low
                    if (Interlocked.CompareExchange(ref m_low, lowLocal + 1, lowLocal) == lowLocal)
                    {
                        //if the specified value is not available (this spot is taken by a push operation, 
                        // but the value is not written into yet), then spin
                        SpinWait spinLocal = new SpinWait(); 
                        while (m_state[lowLocal] == 0) 
                        {
                            spinLocal.SpinOnce(); 
                        }
                        result = m_array[lowLocal];

                        //if the current thread sets m_low to SEGMENT_SIZE, which means the current segment becomes 
                        //disposable, then this thread is responsible to dispose this segment, and reset m_head
                        if (lowLocal + 1 >= SEGMENT_SIZE) 
                        { 
                            //  Invariant: we only dispose the current m_head, not any other segment
                            //  In usual situation, disposing a segment is simply seting m_head to m_head.m_next 
                            //  But there is one special case, where m_head and m_tail points to the same and ONLY
                            //segment of the queue: Another thread A is doing Enqueue and finds that it needs to grow,
                            //while the *current* thread is doing *this* Dequeue operation, and finds that it needs to
                            //dispose the current (and ONLY) segment. Then we need to wait till thread A finishes its 
                            //Grow operation, this is the reason of having the following while loop
                            spinLocal = new SpinWait(); 
                            while (m_next == null) 
                            {
                                spinLocal.SpinOnce(); 
                            }
                            Contract.Assert(head == this);
                            head = m_next;
                        } 

                        return true; 
                    } 
                    else
                    { 
                        //CAS failed due to contention: spin briefly and retry
                        spin.SpinOnce();
                        lowLocal = Low; highLocal = High;
                    } 
                }//end of while
                result = default(T); 
                return false; 
            }
 
            /// 
            /// try to peek the current segment
            /// 
            /// holds the return value of the element at the head position, 
            /// value set to default(T) if there is no such an element
            /// true if there are elements in the current segment, false otherwise 
            internal bool TryPeek(out T result) 
            {
                result = default(T); 
                int lowLocal = Low;
                if (lowLocal > High)
                    return false;
                SpinWait spin = new SpinWait(); 
                while (m_state[lowLocal] == 0)
                { 
                    spin.SpinOnce(); 
                }
                result = m_array[lowLocal]; 
                return true;
            }

            ///  
            /// Convert part or all of the current segment into a List
            ///  
            /// the start position 
            /// the end position
            /// the result list  
            internal List ToList(int start, int end)
            {
                List list = new List();
                for (int i = start; i <= end; i++) 
                {
                    SpinWait spin = new SpinWait(); 
                    while (m_state[i] == 0) 
                    {
                        spin.SpinOnce(); 
                    }
                    list.Add(m_array[i]);
                }
                return list; 
            }
 
            ///  
            /// return the position of the head of the current segment
            ///  
            internal int Low
            {
                get
                { 
                    return Math.Min(m_low, SEGMENT_SIZE);
                } 
            } 

            ///  
            /// return the logical position of the tail of the current segment
            /// 
            internal int High
            { 
                get
                { 
                    //if m_high > SEGMENT_SIZE, it means it's out of range, we should return 
                    //SEGMENT_SIZE-1 as the logical position
                    return Math.Min(m_high, SEGMENT_SIZE - 1); 
                }
            }

        } 
    }//end of class Segment
} 

// File provided for Reference Use Only by Microsoft Corporation (c) 2007.
// Copyright (c) Microsoft Corporation. All rights reserved.
#pragma warning disable 0420 

// ==++==
//
//   Copyright (c) Microsoft Corporation.  All rights reserved. 
//
// ==--== 
// =+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+ 
//
// ConcurrentQueue.cs 
//
// [....]
//
// A lock-free, concurrent queue primitive, and its associated debugger view type. 
//
// =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=- 
 
using System;
using System.Collections; 
using System.Collections.Generic;
using System.Diagnostics;
using System.Diagnostics.Contracts;
using System.Runtime.ConstrainedExecution; 
using System.Runtime.InteropServices;
using System.Runtime.Serialization; 
using System.Security; 
using System.Security.Permissions;
using System.Threading; 

namespace System.Collections.Concurrent
{
 
    /// 
    /// Represents a thread-safe first-in, first-out collection of objects. 
    ///  
    /// Specifies the type of elements in the queue.
    ///  
    /// All public  and protected members of  are thread-safe and may be used
    /// concurrently from multiple threads.
    /// 
    [ComVisible(false)] 
    [DebuggerDisplay("Count = {Count}")]
    [DebuggerTypeProxy(typeof(SystemCollectionsConcurrent_ProducerConsumerCollectionDebugView<>))] 
    [HostProtection(Synchronization = true, ExternalThreading = true)] 
    [Serializable]
    public class ConcurrentQueue : IProducerConsumerCollection 
    {
        //fields of ConcurrentQueue
        [NonSerialized]
        private volatile Segment m_head; 

        [NonSerialized] 
        private volatile Segment m_tail; 

        private T[] m_serializationArray; // Used for custom serialization. 

        private const int SEGMENT_SIZE = 32;

        ///  
        /// Initializes a new instance of the  class.
        ///  
        public ConcurrentQueue() 
        {
            m_head = m_tail = new Segment(0); 
        }

        /// 
        /// Initializes the contents of the queue from an existing collection. 
        /// 
        /// A collection from which to copy elements. 
        private void InitializeFromCollection(IEnumerable collection) 
        {
            m_head = m_tail = new Segment(0); 
            int index = 0;
            foreach (T element in collection)
            {
                Contract.Assert(index >= 0 && index < SEGMENT_SIZE); 
                m_tail.UnsafeAdd(element);
                index++; 
 
                if (index >= SEGMENT_SIZE)
                { 
                    m_tail = m_tail.UnsafeGrow();
                    index = 0;
                }
            } 
        }
 
        ///  
        /// Initializes a new instance of the 
        /// class that contains elements copied from the specified collection 
        /// 
        /// The collection whose elements are copied to the new .
        /// The  argument is 
        /// null.
        public ConcurrentQueue(IEnumerable collection) 
        { 
            if (collection == null)
            { 
                throw new ArgumentNullException("collection");
            }

            InitializeFromCollection(collection); 
        }
 
        ///  
        /// Get the data array to be serialized
        ///  
        [OnSerializing]
        private void OnSerializing(StreamingContext context)
        {
            // save the data into the serialization array to be saved 
            m_serializationArray = ToArray();
        } 
 
        /// 
        /// Construct the queue from a previously seiralized one 
        /// 
        [OnDeserialized]
        private void OnDeserialized(StreamingContext context)
        { 
            Contract.Assert(m_serializationArray != null);
            InitializeFromCollection(m_serializationArray); 
            m_serializationArray = null; 
        }
 
        /// 
        /// Copies the elements of the  to an , starting at a particular
        ///  index. 
        /// 
        /// The one-dimensional Array that is the 
        /// destination of the elements copied from the 
        /// . The Array must have zero-based indexing. 
        /// The zero-based index in  at which copying
        /// begins.
        ///  is a null reference (Nothing in
        /// Visual Basic). 
        ///  is less than
        /// zero. 
        ///  
        ///  is multidimensional. -or-
        ///  does not have zero-based indexing. -or- 
        ///  is equal to or greater than the length of the 
        /// -or- The number of elements in the source  is
        /// greater than the available space from  to the end of the destination
        /// . -or- The type of the source  cannot be cast automatically to the type of the
        /// destination . 
        ///  
        void ICollection.CopyTo(Array array, int index)
        { 
            // Validate arguments.
            if (array == null)
            {
                throw new ArgumentNullException("array"); 
            }
 
            // We must be careful not to corrupt the array, so we will first accumulate an 
            // internal list of elements that we will then copy to the array. This requires
            // some extra allocation, but is necessary since we don't know up front whether 
            // the array is sufficiently large to hold the stack's contents.
            ((ICollection)ToList()).CopyTo(array, index);
        }
 
        /// 
        /// Gets a value indicating whether access to the  is 
        /// synchronized with the SyncRoot. 
        /// 
        /// true if access to the  is synchronized 
        /// with the SyncRoot; otherwise, false. For , this property always
        /// returns false.
        bool ICollection.IsSynchronized
        { 
            // Gets a value indicating whether access to this collection is synchronized. Always returns
            // false. The reason is subtle. While access is in face thread safe, it's not the case that 
            // locking on the SyncRoot would have prevented concurrent pushes and pops, as this property 
            // would typically indicate; that's because we internally use CAS operations vs. true locks.
            get { return false; } 
        }


        ///  
        /// Gets an object that can be used to synchronize access to the . This property is not supported. 
        ///  
        /// The SyncRoot property is not supported.
        object ICollection.SyncRoot 
        {
            get
            {
                throw new NotSupportedException(Environment.GetResourceString("ConcurrentCollection_SyncRoot_NotSupported")); 
            }
        } 
 
        /// 
        /// Returns an enumerator that iterates through a collection. 
        /// 
        /// An  that can be used to iterate through the collection.
        IEnumerator IEnumerable.GetEnumerator()
        { 
            return ((IEnumerable)this).GetEnumerator();
        } 
 
        /// 
        /// Attempts to add an object to the .
        /// 
        /// The object to add to the . The value can be a null 
        /// reference (Nothing in Visual Basic) for reference types.
        ///  
        /// true if the object was added successfully; otherwise, false. 
        /// For , this operation will always add the object to the
        /// end of the  
        /// and return true.
        bool IProducerConsumerCollection.TryAdd(T item)
        {
            Enqueue(item); 
            return true;
        } 
 
        /// 
        /// Attempts to remove and return an object from the .
        /// 
        /// 
        /// When this method returns, if the operation was successful,  contains the 
        /// object removed. If no object was available to be removed, the value is unspecified.
        ///  
        /// true if an element was removed and returned succesfully; otherwise, false. 
        /// For , this operation will attempt to remove the object
        /// from the beginning of the . 
        /// 
        bool IProducerConsumerCollection.TryTake(out T item)
        {
            return TryDequeue(out item); 
        }
 
        ///  
        /// Gets a value that indicates whether the  is empty.
        ///  
        /// true if the  is empty; otherwise, false.
        /// 
        /// For determining whether the collection contains any items, use of this property is recommended
        /// rather than retrieving the number of items from the  property and comparing it 
        /// to 0.  However, as this collection is intended to be accessed concurrently, it may be the case
        /// that another thread will modify the collection after  returns, thus invalidating 
        /// the result. 
        /// 
        public bool IsEmpty 
        {
            get
            {
                Segment head = m_head; 
                if (!head.IsEmpty)
                    //fast route 1: 
                    //if current head is not empty, then queue is not empty 
                    return false;
                else if (head.Next == null) 
                    //fast route 2:
                    //if current head is empty and it's the last segment
                    //then queue is empty
                    return true; 
                else
                //slow route: 
                //current head is empty and it is NOT the last segment, 
                //it means another thread is growing new segment
                { 
                    SpinWait spin = new SpinWait();
                    while (head.IsEmpty)
                    {
                        if (head.Next == null) 
                            return true;
 
                        spin.SpinOnce(); 
                        head = m_head;
                    } 
                    return false;
                }
            }
        } 

        ///  
        /// Copies the elements stored in the  to a new array. 
        /// 
        /// A new array containing a snapshot of elements copied from the .
        public T[] ToArray()
        {
            return ToList().ToArray(); 
        }
 
        ///  
        /// Copies the  elements to a new . 
        /// 
        /// A new  containing a snapshot of
        /// elements copied from the .
        private List ToList() 
        {
            //store head and tail positions in buffer, 
            Segment head, tail; 
            int headLow, tailHigh;
            GetHeadTailPositions(out head, out tail, out headLow, out tailHigh); 

            if (head == tail)
            {
                return head.ToList(headLow, tailHigh); 
            }
 
            List list = new List(head.ToList(headLow, SEGMENT_SIZE - 1)); 
            Segment curr = head.Next;
            while (curr != tail) 
            {
                list.AddRange(curr.ToList(0, SEGMENT_SIZE - 1));
                curr = curr.Next;
            } 
            //Add tail segment
            list.AddRange(tail.ToList(0, tailHigh)); 
            return list; 
        }
 
        /// 
        /// Store the position of the current head and tail positions.
        /// 
        /// return the head segment 
        /// return the tail segment
        /// return the head offset 
        /// return the tail offset 
        private void GetHeadTailPositions(out Segment head, out Segment tail,
            out int headLow, out int tailHigh) 
        {
            head = m_head;
            tail = m_tail;
            headLow = head.Low; 
            tailHigh = tail.High;
            SpinWait spin = new SpinWait(); 
 
            //we loop until the observed values are stable and sensible.
            //This ensures that any update order by other methods can be tolerated. 
            while (
                //if head and tail changed, retry
                head != m_head || tail != m_tail
                //if low and high pointers, retry 
                || headLow != head.Low || tailHigh != tail.High
                //if head jumps ahead of tail because of concurrent grow and dequeue, retry 
                || head.m_index > tail.m_index) 
            {
                spin.SpinOnce(); 
                head = m_head;
                tail = m_tail;
                headLow = head.Low;
                tailHigh = tail.High; 
            }
        } 
 

        ///  
        /// Gets the number of elements contained in the .
        /// 
        /// The number of elements contained in the .
        ///  
        /// For determining whether the collection contains any items, use of the 
        /// property is recommended rather than retrieving the number of items from the  
        /// property and comparing it to 0. 
        /// 
        public int Count 
        {
            get
            {
                //store head and tail positions in buffer, 
                Segment head, tail;
                int headLow, tailHigh; 
                GetHeadTailPositions(out head, out tail, out headLow, out tailHigh); 

                if (head == tail) 
                {
                    return tailHigh - headLow + 1;
                }
 
                //head segment
                int count = SEGMENT_SIZE - headLow; 
 
                //middle segment(s), if any, are full.
                //We don't deal with overflow to be consistent with the behavior of generic types in CLR. 
                count += SEGMENT_SIZE * ((int)(tail.m_index - head.m_index - 1));

                //tail segment
                count += tailHigh + 1; 

                return count; 
            } 
        }
 

        /// 
        /// Copies the  elements to an existing one-dimensional Array, starting at the specified array index. 
        /// 
        /// The one-dimensional Array that is the 
        /// destination of the elements copied from the 
        /// . The Array must have zero-based
        /// indexing. 
        /// The zero-based index in  at which copying
        /// begins.
        ///  is a null reference (Nothing in
        /// Visual Basic). 
        ///  is less than
        /// zero. 
        ///  is equal to or greater than the 
        /// length of the 
        /// -or- The number of elements in the source  is greater than the 
        /// available space from  to the end of the destination .
        /// 
        public void CopyTo(T[] array, int index) 
        {
            if (array == null) 
            { 
                throw new ArgumentNullException("array");
            } 

            // We must be careful not to corrupt the array, so we will first accumulate an
            // internal list of elements that we will then copy to the array. This requires
            // some extra allocation, but is necessary since we don't know up front whether 
            // the array is sufficiently large to hold the stack's contents.
            ToList().CopyTo(array, index); 
        } 

 
        /// 
        /// Returns an enumerator that iterates through the .
        ///  
        /// An enumerator for the contents of the . 
        ///  
        /// The enumeration represents a moment-in-time snapshot of the contents
        /// of the queue.  It does not reflect any updates to the collection after 
        ///  was called.  The enumerator is safe to use
        /// concurrently with reads from and writes to the queue.
        /// 
        public IEnumerator GetEnumerator() 
        {
            return ToList().GetEnumerator(); 
        } 

        ///  
        /// Adds an object to the end of the .
        /// 
        /// The object to add to the end of the . The value can be a null reference 
        /// (Nothing in Visual Basic) for reference types.
        ///  
        public void Enqueue(T item) 
        {
            SpinWait spin = new SpinWait(); 
            while (true)
            {
                Segment tail = m_tail;
                if (tail.TryAppend(item, ref m_tail)) 
                    return;
                spin.SpinOnce(); 
            } 
        }
 

        /// 
        /// Attempts to remove and return the object at the beginning of the . 
        /// 
        ///  
        /// When this method returns, if the operation was successful,  contains the 
        /// object removed. If no object was available to be removed, the value is unspecified.
        ///  
        /// true if an element was removed and returned from the beggining of the 
        /// succesfully; otherwise, false.
        public bool TryDequeue(out T result) 
        {
            while (!IsEmpty) 
            { 
                Segment head = m_head;
                if (head.TryRemove(out result, ref m_head)) 
                    return true;
                //since method IsEmpty spins, we don't need to spin in the while loop
            }
            result = default(T); 
            return false;
        } 
 
        /// 
        /// Attempts to return an object from the beginning of the  
        /// without removing it.
        /// 
        /// When this method returns,  contains an object from
        /// the beginning of the  or an 
        /// unspecified value if the operation failed.
        /// true if and object was returned successfully; otherwise, false. 
        public bool TryPeek(out T result) 
        {
            while (!IsEmpty) 
            {
                Segment head = m_head;
                if (head.TryPeek(out result))
                    return true; 
                //since method IsEmpty spins, we don't need to spin in the while loop
            } 
            result = default(T); 
            return false;
        } 


        /// 
        /// private class for ConcurrentQueue. 
        /// a queue is a linked list of small arrays, each node is called a segment.
        /// A segment contains an array, a pointer to the next segment, and m_low, m_high indices recording 
        /// the first and last valid elements of the array. 
        /// 
        private class Segment 
        {
            //we define two volatile arrays: m_array and m_state. Note that the accesses to the array items
            //do not get volatile treatment. But we don't need to worry about loading adjacent elements or
            //store/load on adjacent elements would suffer reordering. 
            // - Two stores:  these are at risk, but CLRv2 memory model guarantees store-release hence we are safe.
            // - Two loads: because one item from two volatile arrays are accessed, the loads of the array references 
            //          are sufficient to prevent reordering of the loads of the elements. 
            internal volatile T[] m_array;
 
            //Each array entry has 2 possible states
            // 0 -- initial
            // 1 -- contains value: it may be dequeued, but value is still accessible
            // todo: change this int array to bitmap 
            private volatile int[] m_state;
 
            //pointer to the next segment. null if the current segment is the last segment 
            private volatile Segment m_next;
 
            //We use this zero based index to track how many segments have been created for the queue, and
            //to compute how many active segments are there currently.
            // * The number of currently active segments is : m_tail.m_index - m_head.m_index + 1;
            // * m_index is incremented with every Segment.Grow operation. We use Int64 type, and we can safely 
            //   assume that it never overflows. To overflow, we need to do 2^63 increments, even at a rate of 4
            //   billion (2^32) increments per second, it takes 2^31 seconds, which is about 64 years. 
            internal readonly long m_index; 

            //indices of where the first and last valid values 
            // - m_low points to the position of the next element to pop from this segment, range [0, infinity)
            //      m_low >= SEGMENT_SIZE implies the segment is disposable
            // - m_high points to the position of the latest pushed element, range [-1, infinity)
            //      m_high == -1 implies the segment is new and empty 
            //      m_high >= SEGMENT_SIZE-1 means this segment is ready to grow.
            //        and the thread who sets m_high to SEGMENT_SIZE-1 is responsible to grow the segment 
            // - Math.Min(m_low, SEGMENT_SIZE) > Math.Min(m_high, SEGMENT_SIZE-1) implies segment is empty 
            // - initially m_low =0 and m_high=-1;
            private volatile int m_low; 
            private volatile int m_high;

            /// 
            /// Create and initialize a segment with the specified index. 
            /// 
            internal Segment(long index) 
            { 
                m_array = new T[SEGMENT_SIZE];
                m_state = new int[SEGMENT_SIZE]; //all initialized to 0 
                m_high = -1;
                Contract.Assert(index >= 0);
                m_index = index;
            } 

            ///  
            /// return the next segment 
            /// 
            internal Segment Next 
            {
                get { return m_next; }
            }
 

            ///  
            /// return true if the current segment is empty (doesn't have any element available to dequeue, 
            /// false otherwise
            ///  
            internal bool IsEmpty
            {
                get { return (Low > High); }
            } 

            ///  
            /// Add an element to the tail of the current segment 
            /// exclusively called by ConcurrentQueue.InitializedFromCollection
            /// InitializeFromCollection is responsible to guaratee that there is no index overflow, 
            /// and there is no contention
            /// 
            /// 
            internal void UnsafeAdd(T value) 
            {
                Contract.Assert(m_high < SEGMENT_SIZE - 1); 
                m_high++; 
                m_array[m_high] = value;
                m_state[m_high] = 1; 
            }

            /// 
            /// Create a new segment and append to the current one 
            /// Does not update the m_tail pointer
            /// exclusively called by ConcurrentQueue.InitializedFromCollection 
            /// InitializeFromCollection is responsible to guaratee that there is no index overflow, 
            /// and there is no contention
            ///  
            /// the reference to the new Segment
            internal Segment UnsafeGrow()
            {
                Contract.Assert(m_high >= SEGMENT_SIZE - 1); 
                Segment newSegment = new Segment(m_index + 1); //m_index is Int64, we don't need to worry about overflow
                m_next = newSegment; 
                return newSegment; 
            }
 
            /// 
            /// Create a new segment and append to the current one
            /// Update the m_tail pointer
            /// This method is called when there is no contention 
            /// 
            internal void Grow(ref Segment tail) 
            { 
                //no CAS is needed, since there is no contention (other threads are blocked, busy waiting)
                Segment newSegment = new Segment(m_index + 1);  //m_index is Int64, we don't need to worry about overflow 
                m_next = newSegment;
                Contract.Assert(tail == this);
                tail = m_next;
            } 

 
            ///  
            /// Try to append an element at the end of this segment.
            ///  
            /// the element to append
            /// The tail.
            /// true if the element is appended, false if the current segment is full
            /// if appending the specified element succeeds, and after which the segment is full, 
            /// then grow the segment
            internal bool TryAppend(T value, ref Segment tail) 
            { 
                //quickly check if m_high is already over the boundary, if so, bail out
                if (m_high >= SEGMENT_SIZE - 1) 
                {
                    return false;
                }
 
                //Now we will use a CAS to increment m_high, and store the result in newhigh.
                //Depending on how many free spots left in this segment and how many threads are doing this Increment 
                //at this time, the returning "newhigh" can be 
                // 1) < SEGMENT_SIZE - 1 : we took a spot in this segment, and not the last one, just insert the value
                // 2) == SEGMENT_SIZE - 1 : we took the last spot, insert the value AND grow the segment 
                // 3) > SEGMENT_SIZE - 1 : we failed to reserve a spot in this segment, we return false to
                //    Queue.Enqueue method, telling it to try again in the next segment.

                int newhigh = SEGMENT_SIZE; //initial value set to be over the boundary 

                //We need do Interlocked.Increment and value/state update in a finally block to ensure that they run 
                //without interuption. This is to prevent anything from happening between them, and another dequeue 
                //thread maybe spinning forever to wait for m_state[] to be 1;
                try 
                { }
                finally
                {
                    newhigh = Interlocked.Increment(ref m_high); 
                    if (newhigh <= SEGMENT_SIZE - 1)
                    { 
                        m_array[newhigh] = value; 
                        m_state[newhigh] = 1;
                    } 

                    //if this thread takes up the last slot in the segment, then this thread is responsible
                    //to grow a new segment. Calling Grow must be in the finally block too for reliability reason:
                    //if thread abort during Grow, other threads will be left busy spinning forever. 
                    if (newhigh == SEGMENT_SIZE - 1)
                    { 
                        Grow(ref tail); 
                    }
                } 

                //if newhigh <= SEGMENT_SIZE-1, it means the current thread successfully takes up a spot
                return newhigh <= SEGMENT_SIZE - 1;
            } 

 
            ///  
            /// try to remove an element from the head of current segment
            ///  
            /// The result.
            /// The head.
            /// return false only if the current segment is empty
            internal bool TryRemove(out T result, ref Segment head) 
            {
                SpinWait spin = new SpinWait(); 
                int lowLocal = Low, highLocal = High; 
                while (lowLocal <= highLocal)
                { 
                    //try to update m_low
                    if (Interlocked.CompareExchange(ref m_low, lowLocal + 1, lowLocal) == lowLocal)
                    {
                        //if the specified value is not available (this spot is taken by a push operation, 
                        // but the value is not written into yet), then spin
                        SpinWait spinLocal = new SpinWait(); 
                        while (m_state[lowLocal] == 0) 
                        {
                            spinLocal.SpinOnce(); 
                        }
                        result = m_array[lowLocal];

                        //if the current thread sets m_low to SEGMENT_SIZE, which means the current segment becomes 
                        //disposable, then this thread is responsible to dispose this segment, and reset m_head
                        if (lowLocal + 1 >= SEGMENT_SIZE) 
                        { 
                            //  Invariant: we only dispose the current m_head, not any other segment
                            //  In usual situation, disposing a segment is simply seting m_head to m_head.m_next 
                            //  But there is one special case, where m_head and m_tail points to the same and ONLY
                            //segment of the queue: Another thread A is doing Enqueue and finds that it needs to grow,
                            //while the *current* thread is doing *this* Dequeue operation, and finds that it needs to
                            //dispose the current (and ONLY) segment. Then we need to wait till thread A finishes its 
                            //Grow operation, this is the reason of having the following while loop
                            spinLocal = new SpinWait(); 
                            while (m_next == null) 
                            {
                                spinLocal.SpinOnce(); 
                            }
                            Contract.Assert(head == this);
                            head = m_next;
                        } 

                        return true; 
                    } 
                    else
                    { 
                        //CAS failed due to contention: spin briefly and retry
                        spin.SpinOnce();
                        lowLocal = Low; highLocal = High;
                    } 
                }//end of while
                result = default(T); 
                return false; 
            }
 
            /// 
            /// try to peek the current segment
            /// 
            /// holds the return value of the element at the head position, 
            /// value set to default(T) if there is no such an element
            /// true if there are elements in the current segment, false otherwise 
            internal bool TryPeek(out T result) 
            {
                result = default(T); 
                int lowLocal = Low;
                if (lowLocal > High)
                    return false;
                SpinWait spin = new SpinWait(); 
                while (m_state[lowLocal] == 0)
                { 
                    spin.SpinOnce(); 
                }
                result = m_array[lowLocal]; 
                return true;
            }

            ///  
            /// Convert part or all of the current segment into a List
            ///  
            /// the start position 
            /// the end position
            /// the result list  
            internal List ToList(int start, int end)
            {
                List list = new List();
                for (int i = start; i <= end; i++) 
                {
                    SpinWait spin = new SpinWait(); 
                    while (m_state[i] == 0) 
                    {
                        spin.SpinOnce(); 
                    }
                    list.Add(m_array[i]);
                }
                return list; 
            }
 
            ///  
            /// return the position of the head of the current segment
            ///  
            internal int Low
            {
                get
                { 
                    return Math.Min(m_low, SEGMENT_SIZE);
                } 
            } 

            ///  
            /// return the logical position of the tail of the current segment
            /// 
            internal int High
            { 
                get
                { 
                    //if m_high > SEGMENT_SIZE, it means it's out of range, we should return 
                    //SEGMENT_SIZE-1 as the logical position
                    return Math.Min(m_high, SEGMENT_SIZE - 1); 
                }
            }

        } 
    }//end of class Segment
} 

// File provided for Reference Use Only by Microsoft Corporation (c) 2007.
// Copyright (c) Microsoft Corporation. All rights reserved.
                        

Link Menu

Network programming in C#, Network Programming in VB.NET, Network Programming in .NET
This book is available now!
Buy at Amazon US or
Buy at Amazon UK