¿Cómo funciona el patrón disruptor de LMAX?

Estoy tratando de entender el patrón disruptivo . He visto el video de InfoQ y he intentado leer su periódico. Entiendo que hay un buffer de anillo involucrado, que se inicializa como una matriz extremadamente grande para aprovechar la localidad de caché, eliminar la asignación de nueva memoria.

Parece que hay uno o más enteros atómicos que hacen un seguimiento de las posiciones. Cada 'evento' parece tener una identificación única y su posición en el anillo se encuentra al encontrar su módulo con respecto al tamaño del anillo, etc., etc.

Desafortunadamente, no tengo un sentido intuitivo de cómo funciona. Hice muchas aplicaciones comerciales y estudié el modelo de actor , miré a SEDA, etc.

En su presentación mencionaron que este patrón es básicamente cómo funcionan los enrutadores; Sin embargo, tampoco he encontrado ninguna buena descripción de cómo funcionan los enrutadores.

¿Hay algunos buenos consejos para una mejor explicación?

El proyecto Google Code hace referencia a un documento técnico sobre la implementación del buffer de anillo, sin embargo, es un poco seco, académico y difícil para alguien que quiere aprender cómo funciona. Sin embargo, hay algunas publicaciones de blog que han comenzado a explicar los aspectos internos de una manera más legible. Hay una explicación del buffer de anillo que es el núcleo del patrón disruptor, una descripción de las barreras del consumidor (la parte relacionada con la lectura del disruptor) y alguna información sobre el manejo de múltiples productores disponibles.

La descripción más simple del Disruptor es: es una forma de enviar mensajes entre hilos de la manera más eficiente posible. Se puede usar como una alternativa a una cola, pero también comparte una serie de características con SEDA y Actors.

En comparación con las colas:

El disruptor proporciona la capacidad de pasar un mensaje a otros hilos, despertándolo si es necesario (similar a un BlockingQueue). Sin embargo, hay 3 diferencias distintas.

1. El usuario del Disruptor define cómo se almacenan los mensajes ampliando la clase Entry y proporcionando una fábrica para realizar la preasignación. Esto permite la reutilización de la memoria (copia) o la entrada podría contener una referencia a otro objeto.
2. Poner mensajes en el disruptor es un proceso de 2 fases, primero se reclama una ranura en el búfer en anillo, que proporciona al usuario la entrada que se puede llenar con los datos apropiados. Entonces se debe confirmar la entrada, este enfoque de 2 fases es necesario para permitir el uso flexible de la memoria mencionada anteriormente. Es el commit lo que hace que el mensaje sea visible para los hilos del consumidor.
3. Es responsabilidad del consumidor realizar un seguimiento de los mensajes que se han consumido desde el búfer en anillo. Alejar esta responsabilidad del búfer en anillo ayudó a reducir la cantidad de contención de escritura ya que cada hilo mantiene su propio contador.

En comparación con los actores

El modelo de actor está más cerca del disruptor que la mayoría de los otros modelos de programación, especialmente si utiliza las clases BatchConsumer / BatchHandler que se proporcionan. Estas clases ocultan todas las complejidades de mantener los números de secuencia consumidos y proporcionan un conjunto de devoluciones de llamada simples cuando ocurren eventos importantes. Sin embargo, hay un par de diferencias sutiles.

1. El Disruptor utiliza un modelo de consumidor de 1 subproceso - 1, donde los actores usan un modelo N: M, es decir, puede tener tantos actores como desee y se distribuirán en un número fijo de subprocesos (generalmente 1 por núcleo).
2. La interfaz BatchHandler proporciona una devolución de llamada adicional (y muy importante) onEndOfBatch(). Esto permite a los consumidores lentos, por ejemplo, aquellos que realizan E / S para agrupar eventos juntos para mejorar el rendimiento. Es posible realizar lotes en otros marcos de Actor, sin embargo, dado que casi todos los demás marcos no proporcionan una devolución de llamada al final del lote, debe usar un tiempo de espera para determinar el final del lote, lo que resulta en una latencia deficiente.

En comparación con SEDA

LMAX creó el patrón Disruptor para reemplazar un enfoque basado en SEDA.

1. La principal mejora que proporcionó sobre SEDA fue la capacidad de trabajar en paralelo. Para hacer esto, el Disruptor admite la transmisión múltiple de los mismos mensajes (en el mismo orden) a múltiples consumidores. Esto evita la necesidad de etapas de horquilla en la tubería.
2. También permitimos que los consumidores esperen los resultados de otros consumidores sin tener que poner otra etapa de espera entre ellos. Un consumidor simplemente puede ver el número de secuencia de un consumidor del que depende. Esto evita la necesidad de unir etapas en la tubería.

En comparación con las barreras de memoria

Otra forma de pensarlo es como una barrera de memoria estructurada y ordenada. Donde la barrera del productor forma la barrera de escritura y la barrera del consumidor es la barrera de lectura.

Primero nos gustaría entender el modelo de programación que ofrece.

Hay uno o más escritores. Hay uno o más lectores. Hay una línea de entradas, totalmente ordenada de antiguo a nuevo (en la foto, de izquierda a derecha). Los escritores pueden agregar nuevas entradas en el extremo derecho. Cada lector lee las entradas secuencialmente de izquierda a derecha. Los lectores no pueden leer escritores anteriores, obviamente.

No existe el concepto de eliminación de entrada. Utilizo "lector" en lugar de "consumidor" para evitar que se consuma la imagen de las entradas. Sin embargo, entendemos que las entradas a la izquierda del último lector se vuelven inútiles.

En general, los lectores pueden leer de forma simultánea e independiente. Sin embargo, podemos declarar dependencias entre los lectores. Las dependencias del lector pueden ser un gráfico acíclico arbitrario. Si el lector B depende del lector A, el lector B no puede leer el lector anterior A.

La dependencia del lector surge porque el lector A puede anotar una entrada, y el lector B depende de esa anotación. Por ejemplo, A realiza algunos cálculos en una entrada y almacena el resultado en el campo ade la entrada. A luego sigue adelante, y ahora B puede leer la entrada y el valor de aA almacenado. Si el lector C no depende de A, C no debería intentar leer a.

Este es de hecho un modelo de programación interesante. Independientemente del rendimiento, el modelo solo puede beneficiar a muchas aplicaciones.

Por supuesto, el objetivo principal de LMAX es el rendimiento. Utiliza un anillo de entradas preasignado. El anillo es lo suficientemente grande, pero está limitado para que el sistema no se cargue más allá de la capacidad de diseño. Si el anillo está lleno, los escritores esperarán hasta que los lectores más lentos avancen y hagan espacio.

Los objetos de entrada se asignan previamente y viven para siempre, para reducir el costo de recolección de basura. No insertamos objetos de entrada nuevos ni eliminamos objetos de entrada antiguos, en cambio, un escritor solicita una entrada preexistente, llena sus campos y notifica a los lectores. Esta aparente acción de 2 fases es realmente simplemente una acción atómica.

setNewEntry(EntryPopulator);

interface EntryPopulator{ void populate(Entry existingEntry); }

La preasignación de entradas también significa que las entradas adyacentes (muy probablemente) se ubican en celdas de memoria adyacentes, y debido a que los lectores leen las entradas secuencialmente, esto es importante para utilizar cachés de CPU.

Y muchos esfuerzos para evitar el bloqueo, CAS, incluso la barrera de la memoria (por ejemplo, use una variable de secuencia no volátil si solo hay un escritor)

Para desarrolladores de lectores: diferentes lectores de anotaciones deben escribir en diferentes campos, para evitar disputas de escritura. (En realidad, deberían escribir en diferentes líneas de caché). Un lector de anotaciones no debe tocar nada que puedan leer otros lectores no dependientes. Es por eso que digo que estos lectores anotan entradas, en lugar de modificar entradas.

Martin Fowler ha escrito un artículo sobre LMAX y el patrón disruptivo , The LMAX Architecture , que puede aclararlo más.

De hecho, me tomé el tiempo para estudiar la fuente real, por pura curiosidad, y la idea detrás de esto es bastante simple. La versión más reciente al momento de escribir esta publicación es 3.2.1.

Hay un búfer que almacena eventos preasignados que contendrán los datos para que los consumidores los lean.

El búfer está respaldado por una matriz de banderas (matriz de enteros) de su longitud que describe la disponibilidad de las ranuras del búfer (para más detalles, ver más detalles). Se accede a la matriz como un Java # AtomicIntegerArray, por lo que para el propósito de esta explicación, también puede suponer que es una.

Puede haber cualquier cantidad de productores. Cuando el productor quiere escribir en el búfer, se genera un número largo (como al llamar a AtomicLong # getAndIncrement, el Disruptor en realidad usa su propia implementación, pero funciona de la misma manera). Llamemos a esto generado por mucho tiempo un Id. De manera similar, se genera un ConsumerCallId cuando un consumidor termina leyendo una ranura de un búfer. Se accede al último ConsumerCallId.

(Si hay muchos consumidores, se elige la llamada con el ID más bajo).

Estos identificadores se comparan, y si la diferencia entre los dos es menor que el lado del búfer, el productor puede escribir.

(Si el productorCallId es mayor que el reciente consumerCallId + bufferSize, significa que el búfer está lleno y el productor se ve obligado a esperar en el bus hasta que haya un lugar disponible).

Luego, al productor se le asigna la ranura en el búfer en función de su callId (que es prducerCallId modulo bufferSize, pero dado que el bufferSize siempre es una potencia de 2 (límite impuesto en la creación del búfer), la operación real utilizada es productorCallId & (bufferSize - 1 )). Entonces es libre de modificar el evento en ese espacio.

(El algoritmo real es un poco más complicado, ya que involucra el almacenamiento en caché del ID de consumidor reciente en una referencia atómica separada, para fines de optimización).

Cuando se modificó el evento, el cambio se "publica". Al publicar el espacio respectivo en la matriz de banderas se llena con la bandera actualizada. El valor del indicador es el número del bucle (productorCallId dividido por bufferSize (de nuevo, ya que bufferSize tiene una potencia de 2, la operación real es un desplazamiento a la derecha).

De manera similar, puede haber cualquier número de consumidores. Cada vez que un consumidor quiere acceder al búfer, se genera un consumerCallId (dependiendo de cómo se agregaron los consumidores al disruptor, el atómico utilizado en la generación de id se puede compartir o separar para cada uno de ellos). Este ConsumerCallId se compara con el más reciente producentCallId, y si es menor de los dos, el lector puede progresar.

(Del mismo modo, si el productorCallId es incluso para el consumidorCallId, significa que el búfer está vacío y el consumidor se ve obligado a esperar. La forma de espera se define mediante una Estrategia de espera durante la creación del disruptor).

Para los consumidores individuales (los que tienen su propio generador de identificación), lo siguiente que se verifica es la capacidad de consumo por lotes. Las ranuras en el búfer se examinan en orden desde el respectivo al consumerCallId (el índice se determina de la misma manera que para los productores), hasta el respectivo al RecentCallId reciente.

Se examinan en un bucle comparando el valor del indicador escrito en la matriz de indicadores, con un valor de indicador generado para el consumerCallId. Si las banderas coinciden, significa que los productores que ocupan los espacios han comprometido sus cambios. Si no, el bucle se rompe y se devuelve el changeId comprometido más alto. Las ranuras de ConsumerCallId para recibir en changeId se pueden consumir en lote.

Si un grupo de consumidores lee juntos (los que tienen un generador de identificación compartida), cada uno solo toma un único callId, y solo se comprueba y devuelve la ranura para ese único callId.

De este artículo :

El patrón disruptor es una cola de procesamiento por lotes respaldada por una matriz circular (es decir, el búfer en anillo) llena de objetos de transferencia preasignados que utiliza barreras de memoria para sincronizar productores y consumidores a través de secuencias.

Las barreras de memoria son un poco difíciles de explicar y el blog de Trisha ha hecho el mejor intento en mi opinión con esta publicación: http://mechanitis.blogspot.com/2011/08/dissecting-disruptor-why-its-so-fast. html

Pero si no desea sumergirse en los detalles de bajo nivel, puede saber que las barreras de memoria en Java se implementan a través de la volatilepalabra clave o a través de java.util.concurrent.AtomicLong. Las secuencias de patrones disruptivos son AtomicLongsy se comunican de un lado a otro entre productores y consumidores a través de barreras de memoria en lugar de bloqueos.

Me resulta más fácil entender un concepto a través del código, por lo que el siguiente código es un simple helloworld de CoralQueue , que es una implementación de patrón disruptor realizada por CoralBlocks con la que estoy afiliado. En el siguiente código, puede ver cómo el patrón disruptor implementa el procesamiento por lotes y cómo el buffer de anillo (es decir, la matriz circular) permite una comunicación libre de basura entre dos hilos:

package com.coralblocks.coralqueue.sample.queue;

import com.coralblocks.coralqueue.AtomicQueue;
import com.coralblocks.coralqueue.Queue;
import com.coralblocks.coralqueue.util.MutableLong;

public class Sample {

    public static void main(String[] args) throws InterruptedException {

        final Queue<MutableLong> queue = new AtomicQueue<MutableLong>(1024, MutableLong.class);

        Thread consumer = new Thread() {

            @Override
            public void run() {

                boolean running = true;

                while(running) {
                    long avail;
                    while((avail = queue.availableToPoll()) == 0); // busy spin
                    for(int i = 0; i < avail; i++) {
                        MutableLong ml = queue.poll();
                        if (ml.get() == -1) {
                            running = false;
                        } else {
                            System.out.println(ml.get());
                        }
                    }
                    queue.donePolling();
                }
            }

        };

        consumer.start();

        MutableLong ml;

        for(int i = 0; i < 10; i++) {
            while((ml = queue.nextToDispatch()) == null); // busy spin
            ml.set(System.nanoTime());
            queue.flush();
        }

        // send a message to stop consumer...
        while((ml = queue.nextToDispatch()) == null); // busy spin
        ml.set(-1);
        queue.flush();

        consumer.join(); // wait for the consumer thread to die...
    }
}