PRODUTTORE MULTIM PRODUTTORE MULTIMER CONSUMATORE LOCK-FREE BLOCCE (o addirittura Aggiorna la coda)

Question

Sto cercando documentazione su come scrivere la coda MP / MC per essere priva di blocco o addirittura senza attesa.Sto usando .net 4.0.Trovato un sacco di codice C ++, ma non ho familiarità con i modelli di memoria, quindi c'è una grande possibilità che presenterò alcuni bug durante il porting a C #.

Answer 1

Perché pensi che tu abbia bisogno di una coda priva di lock?Hai provato ad usare ConcurrentQueue<T> , eventualmente racchiuso all'interno di un BlockingCollection<T> ?

Scrivere il codice multi-filettato è difficile.Scrivere il codice senza blocco è ancora più difficile e non dovresti farlo da solo a meno che tu non abbia davvero.

Answer 2

Come opzione da considerare, c'è un algoritmo di The Limited Multiple Producer Multiples Consumer Code di Dmitry Vyukov .Ho portato l'algoritmo a .NET, puoi trovare Le fonti su GitHub .È molto veloce.

L'algoritmo di Enqueue:

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

.

L'algoritmo Dequeue:

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

La mia prima volta sarebbe con ConcurrentQueue<T> ma è possibile astrarre il tuo archivio dei dati dietro un'interfaccia in modo da poter modificare facilmente le implementazioni.Quindi Benchmark tipici scenari e vedere dove si incontrano problemi.Ricorda: l'ottimazione prematura è la radice di tutto il male.Progetta il tuo sistema quindi non è legato a un'implementazione ma a un contratto e quindi puoi ottimizzare le tue implementazioni tutto ciò che desideri.

Ho dato un'occhiata a ConcurrentQueue<T> con ILSPY e sembra essere un'implementazione gratuita di blocco a prima vista - così buona possibilità che sia esattamente ciò che stai cercando.