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 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
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
/// The collection whose elements are copied to the new The
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
/// The one-dimensional Array that is the
/// destination of the elements copied from the
/// Array must have zero-based indexing.
/// The zero-based index in
///
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
/// 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; }
}
///
/// Gets an object that can be used to synchronize access to the
/// 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 ((IEnumerable)this).GetEnumerator();
}
///
/// Attempts to add an object to the
/// The object to add to the true if the object was added successfully; otherwise, false.
/// For
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, true if an element was removed and returned succesfully; otherwise, false.
/// For
bool IProducerConsumerCollection.TryTake(out T item)
{
return TryDequeue(out item);
}
///
/// Gets a value that indicates whether the
/// true if the
///
/// 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;
}
}
}
///
/// Copies the elements stored in the
/// A new array containing a snapshot of elements copied from the
public T[] ToArray()
{
return ToList().ToArray();
}
///
/// Copies the
/// A new
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
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 Array , starting at the specified array index.
///
/// The one-dimensional Array that is the
/// destination of the elements copied from the
/// Array must have zero-based
/// indexing.
/// The zero-based index in
/// 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 IEnumerator GetEnumerator()
{
return ToList().GetEnumerator();
}
///
/// Adds an object to the end of the
/// The object to add to the end of the
/// Attempts to remove and return the object at the beginning of the
///
/// When this method returns, if the operation was successful, 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;
}
///
/// Attempts to return an object from the beginning of the
/// When this method returns, 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
[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
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
/// The collection whose elements are copied to the new The
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
/// The one-dimensional Array that is the
/// destination of the elements copied from the
/// Array must have zero-based indexing.
/// The zero-based index in
///
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
/// 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; }
}
///
/// Gets an object that can be used to synchronize access to the
/// 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 ((IEnumerable)this).GetEnumerator();
}
///
/// Attempts to add an object to the
/// The object to add to the true if the object was added successfully; otherwise, false.
/// For
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, true if an element was removed and returned succesfully; otherwise, false.
/// For
bool IProducerConsumerCollection.TryTake(out T item)
{
return TryDequeue(out item);
}
///
/// Gets a value that indicates whether the
/// true if the
///
/// 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;
}
}
}
///
/// Copies the elements stored in the
/// A new array containing a snapshot of elements copied from the
public T[] ToArray()
{
return ToList().ToArray();
}
///
/// Copies the
/// A new
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
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 Array , starting at the specified array index.
///
/// The one-dimensional Array that is the
/// destination of the elements copied from the
/// Array must have zero-based
/// indexing.
/// The zero-based index in
/// 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 IEnumerator GetEnumerator()
{
return ToList().GetEnumerator();
}
///
/// Adds an object to the end of the
/// The object to add to the end of the
/// Attempts to remove and return the object at the beginning of the
///
/// When this method returns, if the operation was successful, 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;
}
///
/// Attempts to return an object from the beginning of the
/// When this method returns, 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
This book is available now!
Buy at Amazon US or
Buy at Amazon UK
- CroppedBitmap.cs
- WebPartChrome.cs
- UnauthorizedAccessException.cs
- DataGridViewColumnHeaderCell.cs
- OutputCacheSettings.cs
- HttpDebugHandler.cs
- TaskFileService.cs
- DecimalMinMaxAggregationOperator.cs
- DataGridAddNewRow.cs
- TypographyProperties.cs
- ToolBarTray.cs
- UnmanagedHandle.cs
- XmlSchemaInferenceException.cs
- login.cs
- JsonByteArrayDataContract.cs
- DiscoveryClientChannelBase.cs
- DataGridCell.cs
- Parameter.cs
- Pen.cs
- WsrmFault.cs
- EntityDataSourceChangingEventArgs.cs
- Menu.cs
- UnmanagedMemoryStream.cs
- ParserStack.cs
- Table.cs
- SolidColorBrush.cs
- DataTableReaderListener.cs
- OpCopier.cs
- XamlBuildProvider.cs
- PropertyValueChangedEvent.cs
- RepeaterCommandEventArgs.cs
- filewebrequest.cs
- PersistenceProviderElement.cs
- Attachment.cs
- MimeMapping.cs
- PropertyIDSet.cs
- ExpressionBinding.cs
- TypeInitializationException.cs
- SQLDoubleStorage.cs
- FastEncoderWindow.cs
- BitmapEffect.cs
- UInt32Converter.cs
- XmlQualifiedNameTest.cs
- ClientRoleProvider.cs
- AttributeCallbackBuilder.cs
- BamlLocalizableResource.cs
- IfAction.cs
- CommonGetThemePartSize.cs
- SerializationFieldInfo.cs
- XamlLoadErrorInfo.cs
- TraceHandler.cs
- ForeignConstraint.cs
- ArglessEventHandlerProxy.cs
- Dump.cs
- ScriptDescriptor.cs
- RegexInterpreter.cs
- SecuritySessionSecurityTokenProvider.cs
- MarshalByRefObject.cs
- PreloadedPackages.cs
- XmlTextReaderImplHelpers.cs
- SecurityManager.cs
- Transform.cs
- HtmlGenericControl.cs
- CookielessData.cs
- ReadOnlyCollectionBuilder.cs
- StreamHelper.cs
- QilExpression.cs
- ElementHostPropertyMap.cs
- RepeatInfo.cs
- EntityDataSourceSelectingEventArgs.cs
- PolicyLevel.cs
- ParseNumbers.cs
- DesignBindingEditor.cs
- EntityContainerRelationshipSet.cs
- ImageSourceConverter.cs
- AssemblyName.cs
- BrowserCapabilitiesFactoryBase.cs
- DataRowView.cs
- XmlDigitalSignatureProcessor.cs
- ClientOptions.cs
- BinaryObjectWriter.cs
- DataSetUtil.cs
- QuadTree.cs
- RegularExpressionValidator.cs
- DefaultHttpHandler.cs
- SQLBoolean.cs
- PairComparer.cs
- SHA384.cs
- SiteMapProvider.cs
- TCPListener.cs
- ClassicBorderDecorator.cs
- EventProviderWriter.cs
- DesignerSerializationManager.cs
- DesignerCapabilities.cs
- IConvertible.cs
- _SecureChannel.cs
- FillErrorEventArgs.cs
- SessionState.cs
- XPathScanner.cs
- COM2ComponentEditor.cs