다중 생산자 다중 소비자 잠금 무료 (또는 심지어 기다리지 않는) 대기열

Question

MP / MC 대기열을 쓸 수있는 방법에 대한 문서를 검색하거나 잠금이 없거나 대기 중입니다.나는 .NET 4.0을 사용하고 있습니다.많은 C ++ 코드를 발견했지만 메모리 모델에 대해 매우 익숙하지 않으므로 C #.

로 이식하는 동안 몇 가지 버그를 소개 할 수있는 큰 기회가 있습니다.

Answer 1

왜 잠금없는 대기열이 필요하다고 생각합니까? ConcurrentQueue<T> BlockingCollection<T> ?

멀티 스레드 코드를 작성하는 것은 어렵습니다.잠금없는 코드를 작성하는 것이 더 어려워 지므로 정말로해야하지 않으면 자신을 그렇게해서는 안됩니다.

Answer 2

고려할 수있는 옵션으로 Dmitry Vyukov 에 의한 경계된 여러 생산자 다중 소비자 대기열.알고리즘을 .NET으로 이식하면 github 의 소스를 찾을 수 있습니다.그것은 매우 빠릅니다.

kenqueue 알고리즘 :

public bool TryEnqueue(object item)
{
    do
    {
        var buffer = _buffer; // prefetch the buffer pointer
        var pos = _enqueuePos; // fetch the current position where to enqueue the item
        var index = pos & _bufferMask; // precalculate the index in the buffer for that position
        var cell = buffer[index]; // fetch the cell by the index
        // If its sequence wasn't touched by other producers
        // and we can increment the enqueue position
        if (cell.Sequence == pos && Interlocked.CompareExchange(ref _enqueuePos, pos + 1, pos) == pos)
        {
            // write the item we want to enqueue
            Volatile.Write(ref buffer[index].Element, item);
            // bump the sequence
            buffer[index].Sequence = pos + 1;
            return true;
        }

        // If the queue is full we cannot enqueue and just return false
        if (cell.Sequence < pos)
        {
            return false;
        }

        // repeat the process if other producer managed to enqueue before us
    } while (true);
}

.

큐 큐 알고리즘 :

public bool TryDequeue(out object result)
{
    do
    {
        var buffer = _buffer; // prefetch the buffer pointer
        var bufferMask = _bufferMask; // prefetch the buffer mask
        var pos = _dequeuePos; // fetch the current position from where we can dequeue an item
        var index = pos & bufferMask; // precalculate the index in the buffer for that position
        var cell = buffer[index]; // fetch the cell by the index
        // If its sequence was changed by a producer and wasn't changed by other consumers
        // and we can increment the dequeue position
        if (cell.Sequence == pos + 1 && Interlocked.CompareExchange(ref _dequeuePos, pos + 1, pos) == pos)
        {
            // read the item
            result = Volatile.Read(ref cell.Element);
            // update for the next round of the buffer
            buffer[index] = new Cell(pos + bufferMask + 1, null);
            return true;
        }

        // If the queue is empty return false
        if (cell.Sequence < pos + 1)
        {
            result = default(object);
            return false;
        }

        // repeat the process if other consumer managed to dequeue before us
    } while (true);
}

.

Answer 3

내 첫 번째 이동은 ConcurrentQueue<T>와 함께 있지만 구현을 쉽게 변경할 수 있도록 인터페이스 뒤에 데이터 저장소를 추상화 할 수 있습니다.그런 다음 전형적인 시나리오를 벤치마킹하고 문제가있는 위치를 확인하십시오.기억하십시오 : 조기 최적화는 모든 악의 뿌리입니다.시스템을 설계하여 구현에 묶이지 않고 계약서에 연결 한 다음 구현을 최적화 할 수 있습니다.

iLspy가있는 ConcurrentQueue<T>를 살펴보고 첫눈에 자물쇠가없는 구현 인 것처럼 보입니다. 그래서 좋은 기회가 정확히 당신이 찾고있는 것입니다.