solicitud de revisión: Hilo cola de seguro (paralelo empuje emergente)

Question

Esta es una implementación de una cola segura hilo. El empuje y el pop no se bloquean mutuamente. Sin embargo, el pop va a esperar hasta que se pulse un elemento, si la cola está vacía. Múltiples productores y consumidores pueden utilizar la cola. Por favor, dígame si hay problemas.

Actualización: Editado según respuestas 1) Resuelto el tema de la situación "de cola llena". 2) Hay BlockingCollection<T> y ConcurrentQueue<T> en .NET4. Así que no hay necesidad de reinventar la rueda (para .NET4)

public class CustomQueue<T> where T: class
{
    class Node
    {
        public Node()
        {
            Value = null;
            NextNode = null;
        }

        public Node(T value)
        {
            Value = value;
            NextNode = null;
        }

        public T Value;
        public Node NextNode;
    }

    object PushLocker = new object();
    object PopLocker = new object();
    Semaphore QueueSemaphore;
    volatile int PushIncrement;
    volatile int PopIncrement;
    int MaxItems;
    Node FirstNode;
    Node LastNode;

    public CustomQueue(int maxItems)
    {
        QueueSemaphore = new Semaphore(0, maxItems);
        MaxItems = maxItems;
        FirstNode = LastNode = new Node();
        PushIncrement = 0;
        PopIncrement = 0;
    }

    public int Size()
    {
        return PushIncrement - PopIncrement;
    }

    public bool Push(T value)
    {
        lock(PushLocker)
        {
            if((Size()) >= MaxItems)
            {
                lock(PopLocker)
                {
                    PushIncrement = PushIncrement - PopIncrement;
                    PopIncrement = 0;
                    return false;
                }
            }
            Node newNode = new Node(value);                
            LastNode.NextNode = newNode;
            LastNode = newNode;
            PushIncrement++;
            QueueSemaphore.Release();
            return true;
        }            
    }

    public T Pop()
    {
        QueueSemaphore.WaitOne();
        lock(PopLocker)
        {
            Node tempFirst = FirstNode;
            Node tempNext = FirstNode.NextNode;
            T value = tempNext.Value;
            tempNext.Value = null;
            FirstNode = tempNext;
            PopIncrement++;
            return value;
        }
    }
}

Answer 1

Se ve como una buena aplicación de un vistazo. El uso de diferentes cerraduras es siempre una señal de alerta para mí, así que tomé una buena mirada a algunos de los casos extremos que implica simultáneamente las llamadas a Pop y Push y parece seguro. Sospecho que probablemente educado a sí mismo en la lista enlazada de implemenation una cola de bloqueo eh? La razón por la que esto es seguro es que para que sólo LastNode de referencia desde Push y FirstNode de Pop de lo contrario todo el truco se desmoronaría.

Lo único que sobresale en mí en este momento es que cuando intenta liberar un recuento de un Semaphore será una excepción si ya está lleno así que sería bueno para protegerse contra eso. ¹ de lo contrario va a terminar con nodos adicionales en la lista enlazada y la cola va 1) tener más que el número máximo de elementos y 2) que conseguirá vivo sin litoral.

Actualización:

Me dio un poco más de este pensamiento. Esa llamada Release en el método Push va a ser muy problemático. Sólo veo una solución posible. Eliminar el parámetro maxItems desde el constructor y permitir que el semáforo para contar hasta Int.MaxValue. De lo contrario, se le va a tener que cambiar radicalmente su enfoque resulta en una implementación que no es donde cerca de lo que tiene actualmente.

¹ _{Creo que se dará cuenta de que esto va a ser más difícil de lo que es posible darse cuenta de inmediato.}

Answer 2

1. Considerar la adición de un segundo constructor nodo:

        public Node(T value)
        {
            Value = value;
        }

a continuación, su código de cliente:

            Node newNode = new Node();
            newNode.Value = value;

puede tratar el valor como un invariante:

            Node newNode = new Node(value);

2. luego hacer su campos públicos:

        public T Value;
        public Node NextNode;

en propiedades de auto:

        public T Value { get; private set; };
        public Node NextNode { get; set; };

Así que usted puede abstraer el uso de la aplicación y añadir la validación, otro tipo de elaboración, etc. después de los hechos con una interrupción mínima de código de cliente.

Answer 3

Si usted está haciendo esto para la auto-educación, grande - de lo contrario BlockingCollection<T> o ConcurrentQueue<T> son buenas alternativas.

Un problema que veo aquí es que no hay manera de Pop interrupción una vez que empieza - asume un objeto está a la espera cuando se despierta. ¿Cómo se desactiva esta abajo en la terminación? Un TryPop que los rendimientos true (con un elemento) o false (si no hay datos) podría ser mejor, entonces usted puede ser señal de los subprocesos en espera de cerrar limpiamente una vez que se vacía la cola.

Answer 4

Ver esto si tiene .Net 4

• http://blogs.msdn.com/b/csharpfaq/archive/2010/08/12/blocking-collection-and-the-producer-consumer-problem.aspx

Answer 5

Como soy un fan de los objetos inmutables, aquí es una alternativa a mi respuesta anterior que consideraría un poco más limpia:

public sealed class CustomQueue<T> where T : class
{
    private readonly object pushLocker = new object();
    private readonly object popLocker = new object();
    private readonly Semaphore queueSemaphore;
    private readonly int maxItems;
    private volatile int pushIncrement;
    private volatile int popIncrement;
    private Node firstNode = new Node();
    private Node lastNode;

    public CustomQueue(int maxItems)
    {
        this.maxItems = maxItems;
        this.lastNode = this.firstNode;
        this.queueSemaphore = new Semaphore(0, this.maxItems);
    }

    public int Size
    {
        get
        {
            return this.pushIncrement - this.popIncrement;
        }
    }

    public bool Push(T value)
    {
        lock (this.pushLocker)
        {
            if (this.Size >= this.maxItems)
            {
                lock (this.popLocker)
                {
                    this.pushIncrement = this.pushIncrement - this.popIncrement;
                    this.popIncrement = 0;
                    return false;
                }
            }

            Node newNode = new Node(value, this.lastNode.NextNode);

            this.lastNode = new Node(this.lastNode.Value, newNode);
            this.firstNode = new Node(null, newNode);
            this.pushIncrement++;
            this.queueSemaphore.Release();
            return true;
        }
    }

    public T Pop()
    {
        this.queueSemaphore.WaitOne();
        lock (this.popLocker)
        {
            Node tempNext = this.firstNode.NextNode;
            T value = tempNext.Value;

            this.firstNode = tempNext;
            this.popIncrement++;
            return value;
        }
    }

    private sealed class Node
    {
        private readonly T value;

        private readonly Node nextNode;

        public Node()
        {
        }

        public Node(T value, Node nextNode)
        {
            this.value = value;
            this.nextNode = nextNode;
        }

        public T Value
        {
            get
            {
                return this.value;
            }
        }

        public Node NextNode
        {
            get
            {
                return this.nextNode;
            }
        }
    }
}