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 [ComVisible(false)] [DebuggerDisplay("Count = {Count}")] [DebuggerTypeProxy(typeof(SystemCollectionsConcurrent_ProducerConsumerCollectionDebugView<>))] [HostProtection(Synchronization = true, ExternalThreading = true)] [Serializable] public class ConcurrentQueueare thread-safe and may be used /// concurrently from multiple threads. /// : 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 public ConcurrentQueue() { m_head = m_tail = new Segment(0); } ///class. /// /// Initializes the contents of the queue from an existing collection. /// /// A collection from which to copy elements. private void InitializeFromCollection(IEnumerablecollection) { 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 /// The collection whose elements are copied to the new/// class that contains elements copied from the specified collection /// . /// The public ConcurrentQueue(IEnumerableargument is /// null. 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 /// The one-dimensionalto an , starting at a particular /// index. /// Array that is the /// destination of the elements copied from the ///. The Array must have zero-based indexing. /// The zero-based index inat which copying /// begins. /// /// is a null reference (Nothing in /// Visual Basic). /// is less than /// zero. /// 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); } ///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 . /// /// Gets a value indicating whether access to the ///is /// synchronized with the SyncRoot. /// true if access to the 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; } } ///is synchronized /// with the SyncRoot; otherwise, false. For , this property always /// returns 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 IEnumerator IEnumerable.GetEnumerator() { return ((IEnumerablethat can be used to iterate through the collection. )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 bool IProducerConsumerCollection, this operation will always add the object to the /// end of the /// and return true. .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 bool IProducerConsumerCollection, this operation will attempt to remove the object /// from the beginning of the . /// .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 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; } } } ///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. /// /// 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 private Listcontaining a snapshot of /// elements copied from the . 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 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; } } ////// property is recommended rather than retrieving the number of items from the /// property and comparing it to 0. /// /// Copies the /// The one-dimensionalelements to an existing one-dimensional Array , starting at the specified array index. ///Array that is the /// destination of the elements copied from the ///. The Array must have zero-based /// indexing. /// The zero-based index inat which copying /// begins. /// /// is a null reference (Nothing in /// Visual Basic). /// is less than /// zero. 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); } /// 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 . /// /// 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 /// public IEnumeratorwas called. The enumerator is safe to use /// concurrently with reads from and writes to the queue. /// 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 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; } ////// succesfully; otherwise, false. /// Attempts to return an object from the beginning of the /// When this method returns,/// without removing it. /// 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 ListToList(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 [ComVisible(false)] [DebuggerDisplay("Count = {Count}")] [DebuggerTypeProxy(typeof(SystemCollectionsConcurrent_ProducerConsumerCollectionDebugView<>))] [HostProtection(Synchronization = true, ExternalThreading = true)] [Serializable] public class ConcurrentQueueare thread-safe and may be used /// concurrently from multiple threads. /// : 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 public ConcurrentQueue() { m_head = m_tail = new Segment(0); } ///class. /// /// Initializes the contents of the queue from an existing collection. /// /// A collection from which to copy elements. private void InitializeFromCollection(IEnumerablecollection) { 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 /// The collection whose elements are copied to the new/// class that contains elements copied from the specified collection /// . /// The public ConcurrentQueue(IEnumerableargument is /// null. 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 /// The one-dimensionalto an , starting at a particular /// index. /// Array that is the /// destination of the elements copied from the ///. The Array must have zero-based indexing. /// The zero-based index inat which copying /// begins. /// /// is a null reference (Nothing in /// Visual Basic). /// is less than /// zero. /// 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); } ///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 . /// /// Gets a value indicating whether access to the ///is /// synchronized with the SyncRoot. /// true if access to the 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; } } ///is synchronized /// with the SyncRoot; otherwise, false. For , this property always /// returns 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 IEnumerator IEnumerable.GetEnumerator() { return ((IEnumerablethat can be used to iterate through the collection. )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 bool IProducerConsumerCollection, this operation will always add the object to the /// end of the /// and return true. .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 bool IProducerConsumerCollection, this operation will attempt to remove the object /// from the beginning of the . /// .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 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; } } } ///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. /// /// 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 private Listcontaining a snapshot of /// elements copied from the . 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 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; } } ////// property is recommended rather than retrieving the number of items from the /// property and comparing it to 0. /// /// Copies the /// The one-dimensionalelements to an existing one-dimensional Array , starting at the specified array index. ///Array that is the /// destination of the elements copied from the ///. The Array must have zero-based /// indexing. /// The zero-based index inat which copying /// begins. /// /// is a null reference (Nothing in /// Visual Basic). /// is less than /// zero. 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); } /// 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 . /// /// 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 /// public IEnumeratorwas called. The enumerator is safe to use /// concurrently with reads from and writes to the queue. /// 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 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; } ////// succesfully; otherwise, false. /// Attempts to return an object from the beginning of the /// When this method returns,/// without removing it. /// 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 ListToList(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
This book is available now!
Buy at Amazon US or
Buy at Amazon UK
- MetaChildrenColumn.cs
- Deserializer.cs
- StringSorter.cs
- PathStreamGeometryContext.cs
- BufferModeSettings.cs
- PopupEventArgs.cs
- DataGridColumnHeaderItemAutomationPeer.cs
- ServiceDescription.cs
- GridViewDeleteEventArgs.cs
- SqlDataSourceEnumerator.cs
- ToolStripItemEventArgs.cs
- ExpressionWriter.cs
- QilBinary.cs
- TextAdaptor.cs
- TemplateControlCodeDomTreeGenerator.cs
- XmlSchemaSimpleContentRestriction.cs
- grammarelement.cs
- Point3DAnimationBase.cs
- RequestCachePolicy.cs
- MobileControlsSectionHelper.cs
- ListViewItemSelectionChangedEvent.cs
- BatchServiceHost.cs
- XamlReader.cs
- ColumnMapProcessor.cs
- MILUtilities.cs
- BaseTemplateCodeDomTreeGenerator.cs
- ColorConverter.cs
- ResourceReader.cs
- AsymmetricKeyExchangeDeformatter.cs
- WindowVisualStateTracker.cs
- SHA256.cs
- SafeFileMappingHandle.cs
- HttpWebRequestElement.cs
- XmlSchemaComplexContentExtension.cs
- IsolatedStorageException.cs
- VideoDrawing.cs
- State.cs
- CqlBlock.cs
- HttpPostedFile.cs
- VirtualPathData.cs
- BulletChrome.cs
- HostingEnvironmentException.cs
- ToolStripArrowRenderEventArgs.cs
- SecurityDocument.cs
- DesignerSerializerAttribute.cs
- ManifestResourceInfo.cs
- TextViewElement.cs
- TransactionInformation.cs
- UTF7Encoding.cs
- ContractInferenceHelper.cs
- CompositeFontInfo.cs
- dataprotectionpermission.cs
- XappLauncher.cs
- _Rfc2616CacheValidators.cs
- Pkcs7Recipient.cs
- HopperCache.cs
- HierarchicalDataBoundControlAdapter.cs
- EntityCommandDefinition.cs
- ExpressionVisitor.cs
- OleDbWrapper.cs
- TagMapInfo.cs
- InternalBufferOverflowException.cs
- srgsitem.cs
- NominalTypeEliminator.cs
- SerializationAttributes.cs
- CornerRadiusConverter.cs
- Privilege.cs
- OracleSqlParser.cs
- ToolStripSettings.cs
- DataSvcMapFileSerializer.cs
- UiaCoreProviderApi.cs
- ServiceBuildProvider.cs
- ResourceDescriptionAttribute.cs
- Pair.cs
- ProjectionCamera.cs
- OutputCacheProfile.cs
- ProgressBarRenderer.cs
- SingleSelectRootGridEntry.cs
- BuildProviderCollection.cs
- XmlSchemaCompilationSettings.cs
- EtwTrace.cs
- AncestorChangedEventArgs.cs
- WindowsImpersonationContext.cs
- DocumentPageTextView.cs
- XmlEncodedRawTextWriter.cs
- HttpConfigurationContext.cs
- UnmanagedMarshal.cs
- AudioFormatConverter.cs
- TextRunTypographyProperties.cs
- ButtonChrome.cs
- HttpRuntimeSection.cs
- Int32Animation.cs
- SystemColors.cs
- Block.cs
- HierarchicalDataBoundControl.cs
- BidOverLoads.cs
- PersistChildrenAttribute.cs
- ToggleButton.cs
- ElementMarkupObject.cs
- objectresult_tresulttype.cs