Comment fonctionne le modèle de perturbateur de LMAX?

J'essaie de comprendre le schéma des perturbateurs . J'ai regardé la vidéo InfoQ et essayé de lire leur article. Je comprends qu'il y a un tampon en anneau impliqué, qu'il est initialisé comme un tableau extrêmement grand pour tirer parti de la localité du cache, éliminer l'allocation de nouvelle mémoire.

Il semble qu'il y ait un ou plusieurs entiers atomiques qui gardent une trace des positions. Chaque `` événement '' semble avoir un identifiant unique et sa position dans l'anneau est trouvée en trouvant son module par rapport à la taille de l'anneau, etc., etc.

Malheureusement, je n'ai pas une idée intuitive de son fonctionnement. J'ai fait de nombreuses applications de trading et étudié le modèle d'acteur , regardé SEDA, etc.

Dans leur présentation, ils ont mentionné que ce modèle est essentiellement le fonctionnement des routeurs; Cependant, je n'ai pas non plus trouvé de bonnes descriptions du fonctionnement des routeurs.

Y a-t-il de bons conseils pour une meilleure explication?

Le projet Google Code fait référence à un document technique sur la mise en œuvre de la mémoire tampon en anneau, mais il est un peu sec, académique et difficile pour quelqu'un qui veut apprendre comment cela fonctionne. Cependant, certains articles de blog ont commencé à expliquer les éléments internes de manière plus lisible. Il y a une explication du tampon en anneau qui est au cœur du modèle de perturbateur, une description des barrières aux consommateurs (la partie liée à la lecture du perturbateur) et quelques informations sur la gestion de plusieurs producteurs disponibles.

La description la plus simple du Disruptor est: C'est un moyen d'envoyer des messages entre les threads de la manière la plus efficace possible. Il peut être utilisé comme alternative à une file d'attente, mais il partage également un certain nombre de fonctionnalités avec SEDA et les acteurs.

Par rapport aux files d'attente:

Le perturbateur offre la possibilité de passer un message sur un autre thread, le réveillant si nécessaire (similaire à une BlockingQueue). Cependant, il existe 3 différences distinctes.

1. L'utilisateur du Disruptor définit comment les messages sont stockés en étendant la classe Entry et en fournissant une fabrique pour effectuer la préallocation. Cela permet soit la réutilisation de la mémoire (copie), soit l'entrée peut contenir une référence à un autre objet.
2. La mise en place de messages dans le Disruptor est un processus en deux phases, tout d'abord un créneau est revendiqué dans le tampon en anneau, qui fournit à l'utilisateur l'entrée qui peut être remplie avec les données appropriées. Ensuite, l'entrée doit être validée, cette approche en 2 phases est nécessaire pour permettre l'utilisation flexible de la mémoire mentionnée ci-dessus. C'est la validation qui rend le message visible aux threads de consommation.
3. Il est de la responsabilité du consommateur de garder une trace des messages qui ont été consommés à partir du tampon en anneau. Éloigner cette responsabilité du tampon en anneau lui-même a permis de réduire le nombre de conflits d'écriture, car chaque thread conserve son propre compteur.

Comparé aux acteurs

Le modèle Actor est plus proche du Disruptor que la plupart des autres modèles de programmation, surtout si vous utilisez les classes BatchConsumer / BatchHandler fournies. Ces classes masquent toutes les complexités de la maintenance des numéros de séquence consommés et fournissent un ensemble de rappels simples lorsque des événements importants se produisent. Cependant, il existe quelques différences subtiles.

1. Le disrupteur utilise un modèle de consommateur 1 thread - 1, où les acteurs utilisent un modèle N: M, c'est-à-dire que vous pouvez avoir autant d'acteurs que vous le souhaitez et ils seront répartis sur un nombre fixe de threads (généralement 1 par cœur).
2. L'interface BatchHandler fournit un rappel supplémentaire (et très important) onEndOfBatch(). Cela permet aux consommateurs lents, par exemple ceux qui font des E / S de regrouper des événements pour améliorer le débit. Il est possible de créer des lots dans d'autres cadres Actor, mais comme presque tous les autres cadres ne fournissent pas de rappel à la fin du lot, vous devez utiliser un délai d'attente pour déterminer la fin du lot, ce qui entraîne une latence médiocre.

Comparé à SEDA

LMAX a construit le modèle Disruptor pour remplacer une approche basée sur SEDA.

1. La principale amélioration apportée par rapport à SEDA était la possibilité de travailler en parallèle. Pour ce faire, le disrupteur prend en charge la multidiffusion des mêmes messages (dans le même ordre) vers plusieurs consommateurs. Cela évite le besoin d'étages de fourche dans le pipeline.
2. Nous permettons également aux consommateurs d'attendre les résultats des autres consommateurs sans avoir à mettre une autre étape de mise en file d'attente entre eux. Un consommateur peut simplement regarder le numéro de séquence d'un consommateur dont il dépend. Cela évite le besoin d'étapes de jointure dans le pipeline.

Comparé aux barrières de mémoire

Une autre façon de penser est comme une barrière de mémoire structurée et ordonnée. Là où la barrière du producteur forme la barrière d'écriture et la barrière du consommateur est la barrière de lecture.

Nous aimerions d'abord comprendre le modèle de programmation qu'il propose.

Il y a un ou plusieurs écrivains. Il y a un ou plusieurs lecteurs. Il y a une ligne d'entrées, totalement ordonnées de l'ancien au nouveau (illustré de gauche à droite). Les rédacteurs peuvent ajouter de nouvelles entrées à l'extrémité droite. Chaque lecteur lit les entrées de manière séquentielle de gauche à droite. Les lecteurs ne peuvent pas lire les anciens écrivains, évidemment.

Il n'y a aucun concept de suppression d'entrée. J'utilise "lecteur" au lieu de "consommateur" pour éviter que l'image des entrées ne soit consommée. Cependant, nous comprenons que les entrées à gauche du dernier lecteur deviennent inutiles.

Généralement, les lecteurs peuvent lire simultanément et indépendamment. Cependant, nous pouvons déclarer des dépendances entre les lecteurs. Les dépendances du lecteur peuvent être des graphes acycliques arbitraires. Si le lecteur B dépend du lecteur A, le lecteur B ne peut pas lire l'ancien lecteur A.

La dépendance du lecteur se produit car le lecteur A peut annoter une entrée et le lecteur B dépend de cette annotation. Par exemple, A effectue un calcul sur une entrée et stocke le résultat dans le champ ade l'entrée. A puis continuez, et maintenant B peut lire l'entrée et la valeur de aA stockée. Si le lecteur C ne dépend pas de A, C ne doit pas tenter de lire a.

Il s'agit en effet d'un modèle de programmation intéressant. Quelles que soient les performances, le modèle à lui seul peut bénéficier à de nombreuses applications.

Bien sûr, l'objectif principal de LMAX est la performance. Il utilise un anneau d'entrées pré-alloué. L'anneau est suffisamment grand, mais il est délimité afin que le système ne soit pas chargé au-delà de la capacité de conception. Si l'anneau est plein, le ou les auteurs attendront que les lecteurs les plus lents avancent et fassent de la place.

Les objets d'entrée sont pré-alloués et vivent pour toujours, afin de réduire les coûts de collecte des ordures. Nous n'insérons pas de nouveaux objets d'entrée ni ne supprimons les anciens objets d'entrée, mais un rédacteur demande une entrée préexistante, remplit ses champs et informe les lecteurs. Cette action apparente en 2 phases est vraiment simplement une action atomique

setNewEntry(EntryPopulator);

interface EntryPopulator{ void populate(Entry existingEntry); }

La pré-allocation d'entrées signifie également que les entrées adjacentes (très probablement) se trouvent dans les cellules de mémoire adjacentes, et parce que les lecteurs lisent les entrées de manière séquentielle, il est important d'utiliser les caches CPU.

Et beaucoup d'efforts pour éviter le verrouillage, le CAS, même la barrière de mémoire (par exemple, utilisez une variable de séquence non volatile s'il n'y a qu'un seul écrivain)

Pour les développeurs de lecteurs: différents lecteurs annotés doivent écrire dans des champs différents, pour éviter les conflits d'écriture. (En fait, ils doivent écrire sur différentes lignes de cache.) Un lecteur d'annotation ne doit rien toucher à ce que d'autres lecteurs non dépendants peuvent lire. C'est pourquoi je dis que ces lecteurs annotent les entrées, au lieu de modifier les entrées.

Martin Fowler a écrit un article sur LMAX et le modèle de perturbateur, L'architecture LMAX , qui pourrait le clarifier davantage.

En fait, j'ai pris le temps d'étudier la source réelle, par pure curiosité, et l'idée derrière cela est assez simple. La version la plus récente au moment de la rédaction de cet article est la 3.2.1.

Il existe un tampon stockant des événements pré-alloués qui contiendront les données à lire par les consommateurs.

Le tampon est soutenu par un tableau de drapeaux (tableau d'entiers) de sa longueur qui décrit la disponibilité des emplacements de tampon (voir plus loin pour plus de détails). Le tableau est accessible comme un java # AtomicIntegerArray, donc pour les besoins de cette explication, vous pouvez aussi supposer qu'il en est un.

Il peut y avoir n'importe quel nombre de producteurs. Lorsque le producteur souhaite écrire dans le tampon, un nombre long est généré (comme lors de l'appel à AtomicLong # getAndIncrement, le disrupteur utilise en fait sa propre implémentation, mais il fonctionne de la même manière). Appelons cela généré longtemps un producteurCallId. De manière similaire, un consumerCallId est généré lorsqu'un consommateur ENDS lit un slot dans un tampon. Le dernier ConsumerCallId est accessible.

(S'il y a beaucoup de consommateurs, l'appel avec l'ID le plus bas est choisi.)

Ces identifiants sont ensuite comparés, et si la différence entre les deux est moindre que le côté tampon, le producteur est autorisé à écrire.

(Si le producteurCallId est supérieur au récent consumerCallId + bufferSize, cela signifie que le tampon est plein et que le producteur est obligé d'attendre le bus jusqu'à ce qu'un spot soit disponible.)

Le producteur se voit alors attribuer l'emplacement dans le tampon en fonction de son callId (qui est prducerCallId modulo bufferSize, mais comme le bufferSize est toujours une puissance de 2 (limite appliquée à la création du tampon), l'opération actuall utilisée est le producteurCallId & (bufferSize - 1 )). Il est alors libre de modifier l'événement dans cet emplacement.

(L'algorithme réel est un peu plus compliqué, impliquant la mise en cache du consumerId récent dans une référence atomique distincte, à des fins d'optimisation.)

Lorsque l'événement a été modifié, le changement est "publié". Lors de la publication, l'emplacement respectif dans le tableau d'indicateurs est rempli avec l'indicateur mis à jour. La valeur de l'indicateur est le numéro de la boucle (producteurCallId divisé par bufferSize (encore une fois, puisque bufferSize est une puissance de 2, l'opération réelle est un décalage vers la droite).

De la même manière, il peut y avoir n'importe quel nombre de consommateurs. Chaque fois qu'un consommateur souhaite accéder au tampon, un consumerCallId est généré (en fonction de la façon dont les consommateurs ont été ajoutés au perturbateur, l'atomique utilisé dans la génération d'ID peut être partagé ou séparé pour chacun d'eux). Ce consumerCallId est ensuite comparé au plus récent producentCallId, et s'il est moindre des deux, le lecteur est autorisé à progresser.

(De même, si le producteurCallId est égal au consumerCallId, cela signifie que le tampon est vide et que le consommateur est obligé d'attendre. Le mode d'attente est défini par une WaitStrategy lors de la création du perturbateur.)

Pour les consommateurs individuels (ceux qui ont leur propre générateur d'ID), la prochaine chose vérifiée est la capacité de consommer par lots. Les créneaux dans le tampon sont examinés dans l'ordre de celui respectif au consumerCallId (l'indice est déterminé de la même manière que pour les producteurs), à celui respectif au producteurCallId récent.

Ils sont examinés en boucle en comparant la valeur d'indicateur écrite dans le tableau d'indicateurs, à une valeur d'indicateur générée pour le consumerCallId. Si les drapeaux correspondent, cela signifie que les producteurs remplissant les créneaux ont validé leurs modifications. Sinon, la boucle est rompue et le changeId engagé le plus élevé est renvoyé. Les emplacements de ConsumerCallId à reçus dans changeId peuvent être consommés par lot.

Si un groupe de consommateurs lisent ensemble (ceux avec un générateur d'ID partagé), chacun ne prend qu'un seul callId, et seul l'emplacement pour ce callId unique est vérifié et renvoyé.

De cet article :

Le modèle de perturbateur est une file d'attente de traitement par lots sauvegardée par un tableau circulaire (c'est-à-dire le tampon en anneau) rempli d'objets de transfert pré-alloués qui utilise des barrières de mémoire pour synchroniser les producteurs et les consommateurs à travers des séquences.

Les barrières de mémoire sont un peu difficiles à expliquer et le blog de Trisha a fait de mon mieux avec cet article: http://mechanitis.blogspot.com/2011/08/dissecting-disruptor-why-its-so-fast. html

Mais si vous ne voulez pas plonger dans les détails de bas niveau, vous pouvez simplement savoir que les barrières mémoire en Java sont implémentées via le volatilemot - clé ou via le java.util.concurrent.AtomicLong. Les séquences de motifs perturbateurs sont AtomicLongs et sont communiquées dans les deux sens entre producteurs et consommateurs via des barrières de mémoire au lieu de verrous.

Je trouve plus facile de comprendre un concept grâce au code, donc le code ci-dessous est un simple helloworld de CoralQueue , qui est une implémentation de modèle de perturbateur effectuée par CoralBlocks auquel je suis affilié. Dans le code ci-dessous, vous pouvez voir comment le modèle de perturbateur implémente le traitement par lots et comment le tampon en anneau (c'est-à-dire un tableau circulaire) permet une communication sans gaspillage entre deux 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...
    }
}