저장 구조 설계

C#의 동시성 큐는 배열과 연결 리스트의 조합으로 구현됩니다. 세그먼트(segment)라는 단위로 데이터를 관리하며, 각 세그먼트는 고정 크기 배열(기본 32개 요소)을 포함합니다. 세그먼트는 단방향 연결 구조로 구성되며, 큐는 헤드(첫 번째 세그먼트)와 테일(마지막 세그먼트) 포인터를 유지합니다.

internal class QueueSegment<T>
{
    internal readonly T[] DataBuffer;
    internal readonly bool[] ValidFlags;
    internal volatile QueueSegment<T> NextSegment;
    internal volatile int StartIndex;
    internal volatile int EndIndex;
    internal readonly long SegmentId;
}

public class ConcurrentQueue<T> : IProducerConsumerCollection<T>
{
    private volatile QueueSegment<T> _headSegment;
    private volatile QueueSegment<T> _tailSegment;
    private const int SegmentCapacity = 32;
}

초기화 과정

기본 생성자는 단일 빈 세그먼트를 생성합니다:

public ConcurrentQueue()
{
    _headSegment = _tailSegment = new QueueSegment<T>(0);
}

세그먼트 초기화 시 배열과 유효성 플래그 배열을 생성하며, 시작 인덱스(0)와 종료 인덱스(-1)를 설정합니다.

데이터 삽입 로직

public void Add(T element)
{
    SpinWait spinner = new SpinWait();
    while (true)
    {
        QueueSegment<T> tail = _tailSegment;
        if (tail.TryAdd(element)) return;
        spinner.SpinOnce();
    }
}

internal bool TryAdd(T item)
{
    if (EndIndex >= SegmentCapacity - 1) return false;
    
    int newIndex = Interlocked.Increment(ref EndIndex);
    if (newIndex <= SegmentCapacity - 1)
    {
        DataBuffer[newIndex] = item;
        ValidFlags[newIndex] = true;
    }
    if (newIndex == SegmentCapacity - 1) ExpandSegment();
    return newIndex <= SegmentCapacity - 1;
}

삽입 실패 시 세그먼트 확장이 완료될 때까지 스핀 대기합니다.

세그먼트 확장

private void ExpandSegment()
{
    QueueSegment<T> newSegment = new QueueSegment<T>(SegmentId + 1);
    NextSegment = newSegment;
    _tailSegment = newSegment;
}

새 세그먼트를 생성하고 기존 테일 세그먼트에 연결합니다.

데이터 추출 로직

public bool TryRemove(out T result)
{
    if (IsEmpty) { result = default; return false; }
    
    QueueSegment<T> head = _headSegment;
    return head.TryExtract(out result);
}

internal bool TryExtract(out T item)
{
    SpinWait spinner = new SpinWait();
    while (StartIndex <= EndIndex)
    {
        int currentStart = StartIndex;
        if (Interlocked.CompareExchange(ref StartIndex, currentStart + 1, currentStart) == currentStart)
        {
            while (!ValidFlags[currentStart]) Thread.Yield();
            item = DataBuffer[currentStart];
            if (SnapshotCount == 0) DataBuffer[currentStart] = default;
            if (currentStart + 1 >= SegmentCapacity) RecycleSegment();
            return true;
        }
        spinner.SpinOnce();
    }
    item = default;
    return false;
}

추출 시 CAS(Compare-and-Swap) 연산으로 스레드 안전성을 보장합니다.

세그먼트 재활용

private void RecycleSegment()
{
    while (_headSegment.NextSegment == null) Thread.SpinWait(1);
    _headSegment = _headSegment.NextSegment;
}

모든 요소가 추출된 세그먼트는 가비지 컬렉션 대상이 됩니다.

요소 개수 계산

public int Count
{
    get
    {
        int total = 0;
        QueueSegment<T> current = _headSegment;
        while (current != null)
        {
            int segmentCount = current.EndIndex - current.StartIndex + 1;
            total += segmentCount > 0 ? segmentCount : 0;
            current = current.NextSegment;
        }
        return total;
    }
}

모든 세그먼트를 순회하며 요소 수를 계산하므로 성능 고려가 필요합니다.

스냅샷 처리

public List<T> ToList()
{
    Interlocked.Increment(ref _snapshotCounter);
    try
    {
        List<T> collection = new List<T>();
        QueueSegment<T> current = _headSegment;
        while (current != null)
        {
            for (int i = current.StartIndex; i <= current.EndIndex; i++)
            {
                if (current.ValidFlags[i]) collection.Add(current.DataBuffer[i]);
            }
            current = current.NextSegment;
        }
        return collection;
    }
    finally { Interlocked.Decrement(ref _snapshotCounter); }
}