Fork me on GitHub

[一起讀源碼]走進C#并發隊列ConcurrentQueue的內部世界

決定從這篇文章開始,開一個讀源碼系列,不限制平臺語言或工具,任何自己感興趣的都會寫。前幾天碰到一個小問題又讀了一遍ConcurrentQueue的源碼,那就拿C#中比較常用的并發隊列ConcurrentQueue作為開篇來聊一聊它的實現原理。

話不多說,直奔主題。

要提前說明下的是,本文解析的源碼是基于.NET Framework 4.8版本,地址是:https://referencesource.microsoft.com/#mscorlib/system/Collections/Concurrent/ConcurrentQueue.cs
本來是打算用.NET Core版本的,但是找了一下竟然沒找到:https://github.com/dotnet/runtime/tree/master/src/libraries/System.Collections.Concurrent/src/System/Collections/Concurrent
不知道是我找錯位置了還是咋回事,有知道的大佬告知一下。不過我覺得實現原理應該類似吧,后面找到了我對比一下,不同的話再寫一篇來分析。


帶著問題出發

如果是自己實現一個簡單的隊列功能,我們該如何設計它的存儲結構呢?一般來說有這兩種方式:數組或者鏈表,先來簡單分析下。

我們都知道,數組是固定空間的集合,意味著初始化的時候要指定數組大小,但是隊列的長度是隨時變化的,超出數組大小了怎么辦?這時候就必須要對數組進行擴容。問題又來了,擴容要擴多少呢,少了不夠用多了浪費內存空間。與之相反的,鏈表是動態空間類型的數據結構,元素之間通過指針相連,不需要提前分配空間,需要多少分配多少。但隨之而來的問題是,大量的出隊入隊操作伴隨著大量對象的創建銷毀,GC的壓力又變得非常大。
事實上,在C#的普通隊列Queue類型中選擇使用數組進行實現,它實現了一套擴容機制,這里不再詳細描述,有興趣的直接看源碼,比較簡單。

回到主題,要實現一個高性能的線程安全隊列,我們試著回答以下問題:

  • 存儲結構是怎樣的
  • 如何初始化(初始容量給多少比較好?)
  • 常用操作(入隊出隊)如何實現
  • 線程安全是如何保證的

存儲結構

通過源碼可以看到ConcurrentQueue采用了數組+鏈表的組合模式,充分吸收了2種結構的優點。

具體來說,它的總體結構是一個鏈表,鏈表的每個節點是一個包含數組的特殊對象,我們稱之為Segment(段或節,原話是a queue is a linked list of small arrays, each node is called a segment.),它里面的數組是存儲真實數據的地方,容量固定大小是32,每一個Segment有指向下一個Segment的的指針,以此形成鏈表結構。而隊列中維護了2個特殊的指針,他們分別指向隊列的首段(head segment)和尾段(tail segment),他們對入隊和出隊有著重要的作用。用一張圖來解釋隊列的內部結構:
construction

嗯,畫圖畫到這里突然聯想到,搞成雙向鏈表的話是不是就神似B+樹的葉子節點?技術就是這么奇妙~

段的核心定義為:

/// <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;
}

常規操作

先從初始化一個隊列開始看起。

創建隊列實例

與普通Queue不同的是,ConcurrentQueue不再支持初始化時指定隊列大小(capacity),僅僅提供一個無參構造函數和一個IEnumerable<T>參數的構造函數。

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

無參構造函數很簡單,創建了一個Segment實例并把首尾指針都指向它,此時隊列只包含一個Segment,它的索引是0,隊列容量是32。
繼續看一下Segment是如何被初始化的:

/// <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;
}

Segment只提供了一個構造函數,接受的參數分別是隊列索引和隊列實例,它創建了一個長度為32的數組,并創建了與之對應的狀態數組,然后初始化了位置指針(m_low=0,m_high=-1,此時表示一個空的Segment)。
到這里,一個并發隊列就創建好了。

使用集合創建隊列的過程和上面類似,只是多了兩個步驟:入隊和擴容,下面會重點描述這兩部分所以這里不再過多介紹。


元素入隊

先亮出源碼:

/// <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();
    }
}

通過源碼可以看到,入隊操作是在隊尾(m_tail)進行的,它嘗試在最后一個Segment中追加指定的元素,如果成功了就直接返回,失敗的話就自旋等待,直到成功為止。那什么情況下會失敗呢?這就要繼續看看是如何追加元素的:

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;
}

所以,只有當尾段m_tail裝滿的情況下追加元素才會失敗,這時候必須要等待下一個段產生,也就是擴容(細細品一下Grow這個詞真的很妙),自旋就是在等擴容完成才能有地方放數據。而在保存數據的時候,通過原子自增操作保證了同一個位置只會有一個數據被寫入,從而實現了線程安全。

注意:這里的裝滿并不是指數組每個位置都有數據,而是指最后一個位置已被使用。

繼續看一下擴容是怎么一個過程:

/// <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;
}

在普通隊列中,擴容是通過創建一個更大的數組然后把數據拷貝過去實現擴容的,這個操作比較耗時。而在并發隊列中就非常簡單了,首先創建一個新Segment,然后把當前Segment的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;
}

可以看到只有在隊列不為空(IsEmpty==false)的情況下才會嘗試出隊操作,而出隊是在首段上進行操作的。關于如何判斷隊列是否為空總結就一句話:當首段m_head不包含任何數據且沒有下一段的時候隊列才為空,詳細的判斷過程源碼注釋中寫的很清楚,限于篇幅不詳細介紹。

出隊的本質是從首段中移除低位指針所指向的元素,看一下具體實現步驟:

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;
}

首先,只有當前Segment不為空的情況下才嘗試移除元素,否則就直接返回false。然后通過一個原子操作Interlocked.CompareExchange判斷當前低位指針上是否有其他線程同時也在移除,如果有那就進入自旋等待,沒有的話就從這個位置取出元素并把低位指針往前推進一位。如果當前隊列沒有正在進行截取快照的操作,那取出元素后還要把這個位置給釋放掉。當這個Segment的所有元素都被移除掉了,這時候要把它丟棄,簡單來說就是讓隊列的首段指針指向它的下一段即可,丟棄的這一段等著GC來收拾它。

這里稍微提一下Interlocked.CompareExchange,它的意思是比較和交換,也就是更為大家所熟悉的CAS(Compare-and-Swap),它主要做了以下2件事情:

  • 比較m_low和lowLocal的值是否相等
  • 如果相等則m_low=lowLocal+1,如果不相等就什么都不做,不管是否相等,始終返回m_low的原始值

整個操作是原子性的,對CPU而言就是一條指令,這樣就可以保證當前位置只有一個線程執行出隊操作。

還有一個TryPeek()方法和出隊類似,它是從隊首獲取一個元素但是無需移除該元素,可以看做Dequeue的簡化版,不再詳細介紹。


獲取隊列中元素的數量

與普通Queue不同的是,ConcurrentQueue并沒有維護一個表示隊列中元素個數的計數器,那就意味著要得到這個數量必須實時去計算。我們看一下計算過程:

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;
    }
}

大致思路是,先計算(GetHeadTailPositions)出首段的低位指針和尾段的高位指針,這中間的總長度就是我們要的數量,然后分成3節依次累加每一個Segment包含的元素個數得到最終的隊列長度,可以看到這是一個開銷比較大的操作。
正因為如此,微軟官方推薦使用IsEmpty屬性來判斷隊列是否為空,而不是使用隊列長度Count==0來判斷,使用ConcurrentStack也是一樣。


截取快照(take snapshot)

所謂的take snapshot就是指一些格式轉換的操作,例如ToArray()ToList()GetEnumerator()這種類型的方法。在前面隊列的核心定義中我們提到有一個m_numSnapshotTakers字段,這時候就派上用場了。下面以比較典型的ToList()源碼舉例說明:

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;
}

可以看到,ToList的邏輯和Count非常相似,都是先計算出兩個首尾位置指針,然后把隊列分為3節依次遍歷處理,最大的不同之處在于方法的開頭和結尾分別對m_numSnapshotTakers做了一個原子操作。
在方法的第一行,使用Interlocked.Increment做了一次遞增,這時候表示隊列正在進行一次截取快照操作,在處理完后又在finally中用Interlocked.Decrement做了一次遞減表示當前操作已完成,這樣確保了在進行快照時不被出隊影響。感覺這塊很難描述的特別好,所以保留了原始的英文注釋,大家慢慢體會。

到這里,基本把ConcurrentQueue的核心說清楚了。


總結一下

回到文章開頭提出的幾個問題,現在應該有了很清晰的答案:

  • 存儲結構 -- 采用數組和鏈表的組合形式
  • 如何初始化 -- 創建固定大小的段,無需指定初始容量
  • 常用操作如何實現 -- 尾段入隊,首段出隊
  • 線程安全問題 -- 使用SpinWait自旋等待和原子操作實現

以上所述均是個人理解,如果有錯誤的地方還請不吝指正,以免誤導他人。

推薦相關閱讀,篇篇都是干貨:http://www.jsfhjj.com/lucifer1982/category/126755.html


posted @ 2020-04-07 09:02  balahoho  閱讀(...)  評論(...編輯  收藏
返回頂部
最新chease0ldman老人