Cấu trúc lưu trữ

Phiên bản .NET Core của ConcurrentQueue sử dụng mô hình phân đoạn (Segment) kết hợp với hàng đợi. Mỗi phân đoạn đóng vai trò như bộ chứa luồng an toàn, có thể hoạt động độc lập. Khác biệt chính so với Framework là kích thước phân đoạn không cố định (từ 32 đến 1,048,576 phần tử) và loại bỏ chỉ mục dài, tạo thành hàng đợi không giới hạn thực sự.

Cấu trúc phân đoạn:

internal sealed class ConcurrentQueueSegment<T>
{
    internal readonly Bucket[] _buckets;
    internal readonly int _bucketMask;
    internal PaddedHeadAndTail _headAndTail;
    internal bool _preservedForObservation;
    internal bool _frozenForEnqueues;
    internal ConcurrentQueueSegment<T>? _nextSegment;
}

internal struct Bucket
{
    [AllowNull, MaybeNull] public T Element;
    public int SequenceValue;
}

Cấu trúc hàng đợi:

public class ConcurrentQueue<T> : IProducerConsumerCollection<T>, IReadOnlyCollection<T>
{
    private const int MinSegmentSize = 32;
    private const int MaxSegmentSize = 1024 * 1024;
    private readonly object _segmentLock;
    private volatile ConcurrentQueueSegment<T> _tail;
    private volatile ConcurrentQueueSegment<T> _head;
}

Thao tác cơ bản

Khởi tạo

Tạo hàng đợi trống:

public ConcurrentQueue()
{
    _segmentLock = new object();
    _tail = _head = new ConcurrentQueueSegment<T>(MinSegmentSize);
}

Khởi tạo với tập hợp dữ liệu:

public ConcurrentQueue(IEnumerable<T> collection)
{
    int size = MinSegmentSize;
    if (collection is ICollection<T> c)
    {
        int count = c.Count;
        if (count > size)
        {
            size = Math.Min(CalculatePowerOfTwo(count), MaxSegmentSize);
        }
    }
    _tail = _head = new ConcurrentQueueSegment<T>(size);
    foreach (T element in collection) Enqueue(element);
}

Thêm phần tử

public void Enqueue(T element)
{
    if (!_tail.TryEnqueue(element)) EnqueueSlow(element);
}

private void EnqueueSlow(T element)
{
    while (true)
    {
        ConcurrentQueueSegment<T> tail = _tail;
        if (tail.TryEnqueue(element)) return;
        
        lock (_segmentLock)
        {
            if (tail == _tail)
            {
                tail.LockForEnqueues();
                int newSize = tail._preservedForObservation ? 
                    MinSegmentSize : Math.Min(tail.Capacity * 2, MaxSegmentSize);
                var newTail = new ConcurrentQueueSegment<T>(newSize);
                tail._nextSegment = newTail;
                _tail = newTail;
            }
        }
    }
}

public bool TryEnqueue(T element)
{
    Bucket[] buckets = _buckets;
    SpinWait waiter = default;
    while (true)
    {
        int currentTail = Volatile.Read(ref _headAndTail.Tail);
        int bucketIndex = currentTail & _bucketMask;
        int sequence = Volatile.Read(ref buckets[bucketIndex].SequenceValue);
        
        int diff = sequence - currentTail;
        if (diff == 0)
        {
            if (Interlocked.CompareExchange(ref _headAndTail.Tail, currentTail + 1, currentTail) == currentTail)
            {
                buckets[bucketIndex].Element = element;
                Volatile.Write(ref buckets[bucketIndex].SequenceValue, currentTail + 1);
                return true;
            }
        }
        else if (diff < 0) return false;
        waiter.SpinOnce(sleep1Threshold: -1);
    }
}

Lấy phần tử

public bool TryDequeue([MaybeNullWhen(false)] out T element) =>
    _head.TryDequeue(out element) || TryDequeueSlow(out element);

private bool TryDequeueSlow([MaybeNullWhen(false)] out T element)
{
    while (true)
    {
        ConcurrentQueueSegment<T> head = _head;
        if (head.TryDequeue(out element)) return true;
        if (head._nextSegment == null)
        {
            element = default!;
            return false;
        }
        if (head.TryDequeue(out element)) return true;
        
        lock (_segmentLock)
        {
            if (head == _head) _head = head._nextSegment;
        }
    }
}

public bool TryDequeue([MaybeNullWhen(false)] out T element)
{
    Bucket[] buckets = _buckets;
    SpinWait waiter = default;
    while (true)
    {
        int currentHead = Volatile.Read(ref _headAndTail.Head);
        int bucketIndex = currentHead & _bucketMask;
        int sequence = Volatile.Read(ref buckets[bucketIndex].SequenceValue);
        
        int diff = sequence - (currentHead + 1);
        if (diff == 0)
        {
            if (Interlocked.CompareExchange(ref _headAndTail.Head, currentHead + 1, currentHead) == currentHead)
            {
                element = buckets[bucketIndex].Element!;
                if (!Volatile.Read(ref _preservedForObservation))
                {
                    buckets[bucketIndex].Element = default;
                    Volatile.Write(ref buckets[bucketIndex].SequenceValue, currentHead + buckets.Length);
                }
                return true;
            }
        }
        else if (diff < 0)
        {
            bool isLocked = _frozenForEnqueues;
            int currentTail = Volatile.Read(ref _headAndTail.Tail);
            if (currentTail - currentHead <= 0 || (isLocked && (currentTail - FreezeOffset - currentHead <= 0)))
            {
                element = default!;
                return false;
            }
        }
        waiter.SpinOnce(sleep1Threshold: -1);
    }
}

Kiểm tra trống

public bool IsEmpty => !TryPeek(out _, useResult: false);

private bool TryPeek([MaybeNullWhen(false)] out T element, bool useResult)
{
    ConcurrentQueueSegment<T> segment = _head;
    while (true)
    {
        if (segment.TryPeek(out element, useResult)) return true;
        ConcurrentQueueSegment<T> next = Volatile.Read(ref segment._nextSegment);
        if (next != null) segment = next;
        else if (Volatile.Read(ref segment._nextSegment) == null) break;
    }
    element = default!;
    return false;
}

Chuyển đổi dữ liệu

public T[] ToArray()
{
    CaptureState(out ConcurrentQueueSegment<T> head, out int headPos, out ConcurrentQueueSegment<T> tail, out int tailPos);
    long count = CalculateCount(head, headPos, tail, tailPos);
    T[] array = new T[count];
    
    using (IEnumerator<T> enumerator = IterateElements(head, headPos, tail, tailPos))
    {
        int index = 0;
        while (enumerator.MoveNext()) array[index++] = enumerator.Current;
    }
    return array;
}

private void CaptureState(out ConcurrentQueueSegment<T> head, out int headPos, out ConcurrentQueueSegment<T> tail, out int tailPos)
{
    lock (_segmentLock)
    {
        head = _head;
        tail = _tail;
        for (ConcurrentQueueSegment<T> s = head; ; s = s._nextSegment!)
        {
            s._preservedForObservation = true;
            if (s == tail) break;
        }
        tail.LockForEnqueues();
        headPos = Volatile.Read(ref head._headAndTail.Head);
        tailPos = Volatile.Read(ref tail._headAndTail.Tail);
    }
}