[一起读源码]走进C#并发队列ConcurrentQueue的内部世界

/// <summary>
/// private class for ConcurrentQueue.
/// 链表节点（段）
/// </summary>
private class Segment
{
    //实际存储数据的容器
    internal volatile T[] m_array;

    //存储对应位置数据的状态，当数据的对应状态位标记为true时该数据才是有效的
    internal volatile VolatileBool[] m_state;

    //下一段的指针
    private volatile Segment m_next;

    //当前段在队列中的索引
    internal readonly long m_index;

    //两个位置指针
    private volatile int m_low;
    private volatile int m_high;

    //所属的队列实例
    private volatile ConcurrentQueue<T> m_source;
}

/// <summary>
/// 线程安全的先进先出集合，
/// </summary>
public class ConcurrentQueue<T> : IProducerConsumerCollection<T>, IReadOnlyCollection<T>
{
    //首段
    [NonSerialized]
    private volatile Segment m_head;

    //尾段
    [NonSerialized]
    private volatile Segment m_tail;
 
    //每一段的大小
    private const int SEGMENT_SIZE = 32;
 
    //截取快照的操作数量
    [NonSerialized]
    internal volatile int m_numSnapshotTakers = 0;
}

/// <summary>
/// Initializes a new instance of the <see cref="ConcurrentQueue{T}"/> class.
/// </summary>
public ConcurrentQueue()
{
    m_head = m_tail = new Segment(0, this);
}

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

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

internal bool TryAppend(T value)
{
    //先判断一下高位指针有没有达到数组边界（也就是数组是否装满了）
    if (m_high >= SEGMENT_SIZE - 1)
    {
        return false;
    }
    int newhigh = SEGMENT_SIZE;
    try
    { }
    finally
    {
        //使用原子操作让高位指针加1
        newhigh = Interlocked.Increment(ref m_high);
        //如果数组还有空位
        if (newhigh <= SEGMENT_SIZE - 1)
        {
            //把数据放到数组中，同时更新状态
            m_array[newhigh] = value;
            m_state[newhigh].m_value = true;
        }
        //数组满了要触发扩容
        if (newhigh == SEGMENT_SIZE - 1)
        {
            Grow();
        }
    }
    return newhigh <= SEGMENT_SIZE - 1;
}

/// <summary>
/// Create a new segment and append to the current one
/// Update the m_tail pointer
/// This method is called when there is no contention
/// </summary>
internal void Grow()
{
    //no CAS is needed, since there is no contention (other threads are blocked, busy waiting)
    Segment newSegment = new Segment(m_index + 1, m_source);  //m_index is Int64, we don't need to worry about overflow
    m_next = newSegment;
    Contract.Assert(m_source.m_tail == this);
    m_source.m_tail = m_next;
}

public bool TryDequeue(out T result)
{
    while (!IsEmpty)
    {
        Segment head = m_head;
        if (head.TryRemove(out result))
            return true;
        //since method IsEmpty spins, we don't need to spin in the while loop
    }
    result = default(T);
    return false;
}

internal bool TryRemove(out T result)
{
    SpinWait spin = new SpinWait();
    int lowLocal = Low, highLocal = High;
    //判断当前段是否为空
    while (lowLocal <= highLocal)
    {
        //判断低位指针位置是否可以移除
        if (Interlocked.CompareExchange(ref m_low, lowLocal + 1, lowLocal) == lowLocal)
        {
            SpinWait spinLocal = new SpinWait();
            //判断元素是否有效
            while (!m_state[lowLocal].m_value)
            {
                spinLocal.SpinOnce();
            }
            //取出元素
            result = m_array[lowLocal];
            //释放引用关系
            if (m_source.m_numSnapshotTakers <= 0)
            {
                m_array[lowLocal] = default(T);
            }
            //判断当前段的元素是否全部被移除了，要丢弃它
            if (lowLocal + 1 >= SEGMENT_SIZE)
            {
                spinLocal = new SpinWait();
                while (m_next == null)
                {
                    spinLocal.SpinOnce();
                }
                Contract.Assert(m_source.m_head == this);
                m_source.m_head = m_next;
            }
            return true;
        }
        else
        {
            //线程竞争失败，自旋等待并重置
            spin.SpinOnce();
            lowLocal = Low; highLocal = High;
        }
    }//end of while
    result = default(T);
    return false;
}

public int Count
{
    get
    {
        Segment head, tail;
        int headLow, tailHigh;
        GetHeadTailPositions(out head, out tail, out headLow, out tailHigh);

        if (head == tail)
        {
            return tailHigh - headLow + 1;
        }

        int count = SEGMENT_SIZE - headLow;
        count += SEGMENT_SIZE * ((int)(tail.m_index - head.m_index - 1));
        count += tailHigh + 1;

        return count;
    }
}

private List<T> ToList()
{
    // Increments the number of active snapshot takers. This increment must happen before the snapshot is
    // taken. At the same time, Decrement must happen after list copying is over. Only in this way, can it
    // eliminate race condition when Segment.TryRemove() checks whether m_numSnapshotTakers == 0.
    Interlocked.Increment(ref m_numSnapshotTakers);

    List<T> list = new List<T>();
    try
    {
        Segment head, tail;
        int headLow, tailHigh;
        GetHeadTailPositions(out head, out tail, out headLow, out tailHigh);

        if (head == tail)
        {
            head.AddToList(list, headLow, tailHigh);
        }
        else
        {
            head.AddToList(list, headLow, SEGMENT_SIZE - 1);
            Segment curr = head.Next;
            while (curr != tail)
            {
                curr.AddToList(list, 0, SEGMENT_SIZE - 1);
                curr = curr.Next;
            }
            tail.AddToList(list, 0, tailHigh);
        }
    }
    finally
    {
        // This Decrement must happen after copying is over.
        Interlocked.Decrement(ref m_numSnapshotTakers);
    }
    return list;
}