Como funciona o padrão de disruptor do LMAX?

Estou tentando entender o padrão do disruptor . Eu assisti o vídeo da InfoQ e tentei ler o jornal deles. Eu entendo que há um buffer de anel envolvido, que é inicializado como uma matriz extremamente grande para aproveitar a localização do cache, eliminar a alocação de nova memória.

Parece que há um ou mais números atômicos que controlam as posições. Cada 'evento' parece ter um ID único e sua posição no anel é encontrada ao encontrar seu módulo em relação ao tamanho do anel, etc., etc.

Infelizmente, não tenho um senso intuitivo de como isso funciona. Eu fiz muitas aplicações de negociação e estudei o modelo de ator , observei a SEDA, etc.

Em sua apresentação, eles mencionaram que esse padrão é basicamente como os roteadores funcionam; no entanto, também não encontrei boas descrições de como os roteadores funcionam.

Existem boas dicas para uma melhor explicação?

O projeto do Google Code faz referência a um documento técnico sobre a implementação do buffer de anel, no entanto, é um pouco seco, acadêmico e difícil para alguém que quer aprender como funciona. No entanto, existem algumas postagens no blog que começaram a explicar os internos de uma maneira mais legível. Há uma explicação do buffer de anel que é o núcleo do padrão do disruptor, uma descrição das barreiras do consumidor (a parte relacionada à leitura do disruptor) e algumas informações sobre o manuseio de vários produtores disponíveis.

A descrição mais simples do Disruptor é: É uma maneira de enviar mensagens entre threads da maneira mais eficiente possível. Ele pode ser usado como uma alternativa a uma fila, mas também compartilha vários recursos com o SEDA e os atores.

Comparado às filas:

O Disruptor fornece a capacidade de passar uma mensagem para outros threads, ativando-a se necessário (semelhante a um BlockingQueue). No entanto, existem 3 diferenças distintas.

1. O usuário do Disruptor define como as mensagens são armazenadas estendendo a classe Entry e fornecendo uma fábrica para realizar a pré-localização. Isso permite a reutilização da memória (cópia) ou a Entrada pode conter uma referência a outro objeto.
2. Colocar mensagens no disruptor é um processo de duas fases, primeiro um slot é reivindicado no buffer de anel, que fornece ao usuário a entrada que pode ser preenchida com os dados apropriados. Em seguida, a entrada deve ser confirmada; essa abordagem em duas fases é necessária para permitir o uso flexível da memória mencionada acima. É o commit que torna a mensagem visível para os threads do consumidor.
3. É de responsabilidade do consumidor acompanhar as mensagens que foram consumidas no buffer de toque. Afastar essa responsabilidade do próprio buffer de anel ajudou a reduzir a quantidade de contenção de gravação, pois cada thread mantém seu próprio contador.

Comparado com os atores

O modelo Actor está mais próximo do Disruptor do que a maioria dos outros modelos de programação, especialmente se você usar as classes BatchConsumer / BatchHandler fornecidas. Essas classes ocultam todas as complexidades de manter os números de sequência consumidos e fornecem um conjunto de retornos de chamada simples quando ocorrem eventos importantes. No entanto, existem algumas diferenças sutis.

1. O Disruptor usa um modelo de consumidor de 1 thread - 1, onde os atores usam um modelo N: M, ou seja, você pode ter quantos atores quiser e eles serão distribuídos por um número fixo de threads (geralmente 1 por núcleo).
2. A interface BatchHandler fornece um retorno de chamada adicional (e muito importante) onEndOfBatch(). Isso permite consumidores lentos, por exemplo, aqueles que executam E / S para agrupar eventos em lote para melhorar a taxa de transferência. É possível fazer lotes em outras estruturas do Actor, no entanto, como quase todas as outras estruturas não fornecem um retorno de chamada no final do lote, é necessário usar um tempo limite para determinar o final do lote, resultando em baixa latência.

Comparado com SEDA

A LMAX construiu o padrão Disruptor para substituir uma abordagem baseada em SEDA.

1. A principal melhoria que ele forneceu sobre a SEDA foi a capacidade de trabalhar em paralelo. Para fazer isso, o Disruptor suporta a transmissão múltipla das mesmas mensagens (na mesma ordem) para vários consumidores. Isso evita a necessidade de estágios de garfo no pipeline.
2. Também permitimos que os consumidores esperem nos resultados de outros consumidores sem ter que colocar outro estágio de fila entre eles. Um consumidor pode simplesmente assistir ao número de sequência de um consumidor do qual ele depende. Isso evita a necessidade de estágios de junção no pipeline.

Comparado com as barreiras de memória

Outra maneira de pensar sobre isso é como uma barreira de memória estruturada e ordenada. Onde a barreira do produtor forma a barreira de gravação e a barreira do consumidor é a barreira de leitura.

Primeiro, gostaríamos de entender o modelo de programação que ele oferece.

Há um ou mais escritores. Existem um ou mais leitores. Há uma linha de entradas, totalmente ordenadas do antigo para o novo (na imagem da esquerda para a direita). Os escritores podem adicionar novas entradas na extremidade direita. Todo leitor lê entradas seqüencialmente da esquerda para a direita. Os leitores não podem ler escritores do passado, obviamente.

Não há conceito de exclusão de entrada. Eu uso "leitor" em vez de "consumidor" para evitar que a imagem das entradas seja consumida. No entanto, entendemos que as entradas à esquerda do último leitor se tornam inúteis.

Geralmente, os leitores podem ler simultaneamente e de forma independente. No entanto, podemos declarar dependências entre os leitores. As dependências do leitor podem ser um gráfico acíclico arbitrário. Se o leitor B depender do leitor A, o leitor B não poderá ler o leitor A.

A dependência do leitor surge porque o leitor A pode anotar uma entrada e o leitor B depende dessa anotação. Por exemplo, A faz algum cálculo em uma entrada e armazena o resultado no campo ana entrada. A então siga em frente, e agora B pode ler a entrada e o valor de aA armazenado. Se o leitor C não depende de A, C não deve tentar ler a.

Este é realmente um modelo de programação interessante. Independentemente do desempenho, o modelo por si só pode beneficiar muitos aplicativos.

Obviamente, o principal objetivo da LMAX é o desempenho. Ele usa um anel de entradas pré-alocado. O anel é grande o suficiente, mas é delimitado para que o sistema não seja carregado além da capacidade de design. Se o anel estiver cheio, o (s) escritor (es) esperará até os leitores mais lentos avançarem e abrirem espaço.

Os objetos de entrada são pré-alocados e permanecem para sempre, para reduzir o custo da coleta de lixo. Não inserimos novos objetos de entrada ou excluímos objetos de entrada antigos; um escritor solicita uma entrada preexistente, preenche seus campos e notifica os leitores. Essa aparente ação em duas fases é realmente simplesmente uma ação atômica

setNewEntry(EntryPopulator);

interface EntryPopulator{ void populate(Entry existingEntry); }

A pré-alocação de entradas também significa que as entradas adjacentes (muito provavelmente) localizam-se nas células de memória adjacentes e, como os leitores leem as entradas sequencialmente, isso é importante para usar caches de CPU.

E muitos esforços para evitar bloqueio, CAS e até barreira de memória (por exemplo, use uma variável de sequência não volátil se houver apenas um gravador)

Para desenvolvedores de leitores: Diferentes leitores de anotação devem gravar em campos diferentes, para evitar contenção de gravação. (Na verdade, eles devem gravar em diferentes linhas de cache.) Um leitor de anotações não deve tocar em nada que outros leitores não dependentes possam ler. É por isso que digo que esses leitores anotam entradas, em vez de modificar entradas.

Martin Fowler escreveu um artigo sobre o LMAX e o padrão de disruptor, The LMAX Architecture , que pode esclarecer melhor.

Na verdade, dediquei um tempo para estudar a fonte real, por pura curiosidade, e a idéia por trás disso é bastante simples. A versão mais recente no momento da redação deste post é 3.2.1.

Há um buffer que armazena eventos pré-alocados que retêm os dados para os consumidores lerem.

O buffer é apoiado por uma matriz de sinalizadores (matriz inteira) de seu comprimento que descreve a disponibilidade dos slots do buffer (veja mais detalhes). O array é acessado como um java # AtomicIntegerArray, portanto, para o propósito desta explicação, você também pode supor que seja um.

Pode haver qualquer número de produtores. Quando o produtor deseja gravar no buffer, um número longo é gerado (como na chamada AtomicLong # getAndIncrement, o Disruptor realmente usa sua própria implementação, mas funciona da mesma maneira). Vamos chamar isso gerado por muito tempo de producerCallId. De maneira semelhante, um consumerCallId é gerado quando um consumidor TERMINA lendo um slot de um buffer. O consumerCallId mais recente é acessado.

(Se houver muitos consumidores, a chamada com o ID mais baixo será escolhida.)

Esses IDs são comparados e, se a diferença entre os dois for menor que o lado do buffer, o produtor poderá escrever.

(Se o producerCallId for maior que o recente consumerCallId + bufferSize, isso significa que o buffer está cheio e o produtor é forçado a esperar no barramento até que um local fique disponível.)

O produtor recebe o slot no buffer com base em seu callId (que é o módulo bufferSize do prducerCallId, mas, como o bufferSize é sempre uma potência de 2 (limite imposto na criação do buffer), a operação real usada é o producerCallId & (bufferSize - 1 )). É grátis modificar o evento nesse slot.

(O algoritmo real é um pouco mais complicado, envolvendo o cache de consumerId recente em uma referência atômica separada, para fins de otimização.)

Quando o evento foi modificado, a alteração é "publicada". Ao publicar o respectivo slot na matriz de sinalizadores, é preenchido com o sinalizador atualizado. O valor do sinalizador é o número do loop (producerCallId dividido por bufferSize (novamente porque bufferSize é a potência de 2, a operação real é uma mudança à direita).

De maneira semelhante, pode haver qualquer número de consumidores. Sempre que um consumidor deseja acessar o buffer, é gerado um consumerCallId (dependendo de como os consumidores foram adicionados ao disruptor, o atômico usado na geração de ID pode ser compartilhado ou separado para cada um deles). Este consumerCallId é então comparado ao producentCallId mais recente e, se for menor dos dois, o leitor poderá progredir.

(Da mesma forma, se o producerCallId for igual ao consumerCallId, isso significa que o buffer é oitenta e o consumidor é forçado a esperar. A maneira de esperar é definida por um WaitStrategy durante a criação do disruptor.)

Para consumidores individuais (aqueles com seu próprio gerador de identificação), a próxima coisa verificada é a capacidade de consumir em lote. Os slots no buffer são examinados em ordem do respectivo ao consumerCallId (o índice é determinado da mesma maneira que para os produtores), ao do respectivo ao recente producerCallId.

Eles são examinados em um loop comparando o valor do sinalizador gravado na matriz do sinalizador com um valor do sinalizador gerado para o consumerCallId. Se as bandeiras coincidirem, significa que os produtores que preencheram os espaços confirmaram suas alterações. Caso contrário, o loop será interrompido e o mais alto changeId confirmado será retornado. Os slots de ConsumerCallId para recebidos em changeId podem ser consumidos em lote.

Se um grupo de consumidores lê em conjunto (aqueles com gerador de identificação compartilhada), cada um recebe apenas um único callId e apenas o slot desse único callId é verificado e retornado.

Deste artigo :

O padrão do disruptor é uma fila de lotes com backup de uma matriz circular (ou seja, o buffer de anel) preenchida com objetos de transferência pré-alocados, que utiliza barreiras de memória para sincronizar produtores e consumidores por meio de seqüências.

Barreiras à memória são meio difíceis de explicar e o blog de Trisha fez a melhor tentativa na minha opinião com este post: http://mechanitis.blogspot.com/2011/08/dissecting-disruptor-why-its-so-fast. html

Mas se você não quiser se aprofundar nos detalhes de baixo nível, poderá saber apenas que as barreiras de memória em Java são implementadas por meio da volatilepalavra - chave ou do java.util.concurrent.AtomicLong. As seqüências de padrões de disruptor são AtomicLongs e são comunicadas entre produtores e consumidores através de barreiras de memória em vez de bloqueios.

Acho que é mais fácil de entender um conceito através de código, de modo que o código abaixo é um simples helloworld de CoralQueue , que é uma implementação padrão disruptor feito por CoralBlocks com que eu sou afiliado. No código abaixo, você pode ver como o padrão do disruptor implementa o lote e como o buffer de anel (ou seja, matriz circular) permite a comunicação sem lixo entre dois threads:

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...
    }
}