Wie funktioniert das Disruptormuster von LMAX?

Ich versuche das Disruptormuster zu verstehen . Ich habe das InfoQ-Video gesehen und versucht, ihre Zeitung zu lesen. Ich verstehe, dass es sich um einen Ringpuffer handelt, der als extrem großes Array initialisiert wird, um die Cache-Lokalität zu nutzen und die Zuweisung von neuem Speicher zu eliminieren.

Es hört sich so an, als gäbe es eine oder mehrere atomare Ganzzahlen, die die Positionen verfolgen. Jedes 'Ereignis' scheint eine eindeutige ID zu erhalten und seine Position im Ring wird ermittelt, indem sein Modul in Bezug auf die Größe des Rings usw. usw. ermittelt wird.

Leider habe ich keine intuitive Vorstellung davon, wie es funktioniert. Ich habe viele Handelsanwendungen gemacht und das Schauspieler-Modell studiert , mir SEDA angesehen usw.

In ihrer Präsentation erwähnten sie, dass dieses Muster im Grunde die Funktionsweise von Routern ist. Ich habe jedoch auch keine guten Beschreibungen der Funktionsweise von Routern gefunden.

Gibt es einige gute Hinweise auf eine bessere Erklärung?

Das Google Code-Projekt verweist zwar auf ein technisches Dokument zur Implementierung des Ringpuffers, ist jedoch für jemanden, der lernen möchte, wie es funktioniert, etwas trocken, akademisch und schwierig. Es gibt jedoch einige Blog-Beiträge, die begonnen haben, die Interna besser lesbar zu erklären. Es gibt eine Erklärung des Ringpuffers , der den Kern des Disruptormusters bildet, eine Beschreibung der Verbraucherbarrieren (der Teil, der sich auf das Lesen vom Disruptor bezieht) und einige Informationen zum Umgang mit mehreren verfügbaren Herstellern .

Die einfachste Beschreibung des Disruptors lautet: Auf diese Weise können Nachrichten auf möglichst effiziente Weise zwischen Threads gesendet werden. Es kann als Alternative zu einer Warteschlange verwendet werden, teilt jedoch auch eine Reihe von Funktionen mit SEDA und Actors.

Im Vergleich zu Warteschlangen:

Der Disruptor bietet die Möglichkeit, eine Nachricht an andere Threads weiterzuleiten und bei Bedarf zu aktivieren (ähnlich wie bei einer BlockingQueue). Es gibt jedoch 3 verschiedene Unterschiede.

1. Der Benutzer des Disruptors definiert, wie Nachrichten gespeichert werden, indem er die Eintragsklasse erweitert und eine Factory bereitstellt, um die Vorbelegung durchzuführen. Dies ermöglicht entweder die Wiederverwendung des Speichers (Kopieren) oder der Eintrag kann einen Verweis auf ein anderes Objekt enthalten.
2. Das Einfügen von Nachrichten in den Disruptor erfolgt in zwei Phasen. Zunächst wird ein Steckplatz im Ringpuffer beansprucht, der dem Benutzer den Eintrag zur Verfügung stellt, der mit den entsprechenden Daten gefüllt werden kann. Dann muss der Eintrag festgeschrieben werden. Dieser 2-Phasen-Ansatz ist erforderlich, um die oben erwähnte flexible Nutzung des Speichers zu ermöglichen. Es ist das Commit, das die Nachricht für die Consumer-Threads sichtbar macht.
3. Es liegt in der Verantwortung des Verbrauchers, die Nachrichten zu verfolgen, die aus dem Ringpuffer verbraucht wurden. Das Entfernen dieser Verantwortung vom Ringpuffer selbst hat dazu beigetragen, die Anzahl der Schreibkonflikte zu verringern, da jeder Thread seinen eigenen Zähler beibehält.

Im Vergleich zu Schauspielern

Das Actor-Modell ist näher am Disruptor als die meisten anderen Programmiermodelle, insbesondere wenn Sie die bereitgestellten BatchConsumer / BatchHandler-Klassen verwenden. Diese Klassen verbergen alle Komplexitäten der Verwaltung der verbrauchten Sequenznummern und bieten eine Reihe einfacher Rückrufe, wenn wichtige Ereignisse auftreten. Es gibt jedoch einige subtile Unterschiede.

1. Der Disruptor verwendet ein 1-Thread-1-Verbrauchermodell, wobei Akteure ein N: M-Modell verwenden, dh Sie können so viele Akteure haben, wie Sie möchten, und sie werden auf eine feste Anzahl von Threads verteilt (im Allgemeinen 1 pro Kern).
2. Die BatchHandler-Schnittstelle bietet einen zusätzlichen (und sehr wichtigen) Rückruf onEndOfBatch(). Dies ermöglicht langsamen Verbrauchern, z. B. solchen, die E / A ausführen, um Ereignisse zusammenzufassen, um den Durchsatz zu verbessern. Es ist möglich, Batching in anderen Actor-Frameworks durchzuführen. Da jedoch fast alle anderen Frameworks am Ende des Batches keinen Rückruf bereitstellen, müssen Sie eine Zeitüberschreitung verwenden, um das Ende des Batches zu bestimmen, was zu einer schlechten Latenz führt.

Im Vergleich zu SEDA

LMAX hat das Disruptor-Muster erstellt, um einen SEDA-basierten Ansatz zu ersetzen.

1. Die Hauptverbesserung gegenüber SEDA war die Fähigkeit, parallel zu arbeiten. Zu diesem Zweck unterstützt der Disruptor das Multi-Casting derselben Nachrichten (in derselben Reihenfolge) an mehrere Verbraucher. Dies vermeidet die Notwendigkeit von Gabelstufen in der Pipeline.
2. Wir erlauben den Verbrauchern auch, auf die Ergebnisse anderer Verbraucher zu warten, ohne eine weitere Warteschlange zwischen ihnen einlegen zu müssen. Ein Verbraucher kann einfach die Sequenznummer eines Verbrauchers beobachten, von der er abhängig ist. Dies vermeidet die Notwendigkeit von Verbindungsphasen in der Pipeline.

Im Vergleich zu Speicherbarrieren

Eine andere Art, darüber nachzudenken, ist eine strukturierte, geordnete Speicherbarriere. Wo die Produzentenbarriere die Schreibbarriere bildet und die Konsumentenbarriere die Lesebarriere ist.

Zunächst möchten wir das Programmiermodell verstehen, das es bietet.

Es gibt einen oder mehrere Autoren. Es gibt einen oder mehrere Leser. Es gibt eine Reihe von Einträgen, die vollständig von alt nach neu geordnet sind (von links nach rechts abgebildet). Autoren können am rechten Ende neue Einträge hinzufügen. Jeder Leser liest die Einträge nacheinander von links nach rechts. Leser können offensichtlich keine früheren Autoren lesen.

Es gibt kein Konzept für das Löschen von Einträgen. Ich verwende "Leser" anstelle von "Verbraucher", um zu vermeiden, dass das Bild von Einträgen verbraucht wird. Wir verstehen jedoch, dass Einträge links vom letzten Leser unbrauchbar werden.

Im Allgemeinen können Leser gleichzeitig und unabhängig lesen. Wir können jedoch Abhängigkeiten zwischen Lesern deklarieren. Leserabhängigkeiten können beliebige azyklische Graphen sein. Wenn Leser B von Leser A abhängt, kann Leser B nicht an Leser A vorbei lesen.

Die Leserabhängigkeit entsteht, weil Leser A einen Eintrag mit Anmerkungen versehen kann und Leser B von dieser Anmerkung abhängt. Beispielsweise führt A eine Berechnung für einen Eintrag durch und speichert das Ergebnis im Feld aim Eintrag. A geht dann weiter und jetzt kann B den Eintrag lesen und den Wert von aA speichern. Wenn Leser C nicht von A abhängt, sollte C nicht versuchen zu lesen a.

Dies ist in der Tat ein interessantes Programmiermodell. Unabhängig von der Leistung kann das Modell allein vielen Anwendungen zugute kommen.

Das Hauptziel von LMAX ist natürlich die Leistung. Es wird ein vorab zugewiesener Ring von Einträgen verwendet. Der Ring ist groß genug, aber begrenzt, damit das System nicht über die Entwurfskapazität hinaus geladen wird. Wenn der Ring voll ist, warten die Autoren, bis die langsamsten Leser vorrücken und Platz schaffen.

Eintragsobjekte sind vorab zugewiesen und leben für immer, um die Kosten für die Speicherbereinigung zu senken. Wir fügen keine neuen Eintragsobjekte ein oder löschen keine alten Eintragsobjekte. Stattdessen fordert ein Verfasser einen bereits vorhandenen Eintrag an, füllt seine Felder aus und benachrichtigt die Leser. Diese scheinbare 2-Phasen-Aktion ist wirklich einfach eine atomare Aktion

setNewEntry(EntryPopulator);

interface EntryPopulator{ void populate(Entry existingEntry); }

Das Vorzuweisen von Einträgen bedeutet auch, dass benachbarte Einträge (sehr wahrscheinlich) in benachbarten Speicherzellen lokalisiert sind. Da Leser Einträge nacheinander lesen, ist es wichtig, CPU-Caches zu verwenden.

Und viele Anstrengungen, um Sperren, CAS und sogar Speicherbarrieren zu vermeiden (z. B. eine nichtflüchtige Sequenzvariable verwenden, wenn nur ein Writer vorhanden ist)

Für Entwickler von Lesern: Verschiedene kommentierende Leser sollten in verschiedene Felder schreiben, um Schreibkonflikte zu vermeiden. (Eigentlich sollten sie in verschiedene Cache-Zeilen schreiben.) Ein Annotationsleser sollte nichts berühren, was andere nicht abhängige Leser lesen könnten. Aus diesem Grund sage ich, dass diese Leser Einträge kommentieren , anstatt Einträge zu ändern .

Martin Fowler hat einen Artikel über LMAX und das Disruptormuster The LMAX Architecture geschrieben , der dies möglicherweise weiter verdeutlicht.

Ich habe mir aus purer Neugier die Zeit genommen, die eigentliche Quelle zu studieren, und die Idee dahinter ist recht einfach. Die aktuellste Version zum Zeitpunkt des Schreibens dieses Beitrags ist 3.2.1.

Es gibt einen Puffer, in dem vorab zugewiesene Ereignisse gespeichert sind, die die Daten enthalten, die die Verbraucher lesen können.

Der Puffer wird durch ein Array von Flags (Integer-Array) seiner Länge gesichert, das die Verfügbarkeit der Puffersteckplätze beschreibt (Einzelheiten siehe weiter unten). Auf das Array wird wie auf ein Java # AtomicIntegerArray zugegriffen. Für die Zwecke dieser Erläuterung können Sie also auch davon ausgehen, dass es sich um eines handelt.

Es kann eine beliebige Anzahl von Herstellern geben. Wenn der Produzent in den Puffer schreiben möchte, wird eine lange Nummer generiert (wie beim Aufrufen von AtomicLong # getAndIncrement verwendet der Disruptor tatsächlich seine eigene Implementierung, funktioniert jedoch auf die gleiche Weise). Nennen wir dies lange eine ProducerCallId. In ähnlicher Weise wird eine consumerCallId generiert, wenn ein Consumer einen Slot aus einem Puffer liest. Auf die neueste consumerCallId wird zugegriffen.

(Wenn es viele Verbraucher gibt, wird der Anruf mit der niedrigsten ID ausgewählt.)

Diese IDs werden dann verglichen, und wenn der Unterschied zwischen den beiden geringer ist als die Pufferseite, darf der Produzent schreiben.

(Wenn die ProducerCallId größer als die aktuelle ConsumerCallId + bufferSize ist, bedeutet dies, dass der Puffer voll ist und der Produzent gezwungen ist, mit dem Bus zu warten, bis ein Spot verfügbar ist.)

Dem Produzenten wird dann der Steckplatz im Puffer basierend auf seiner callId (prducerCallId modulo bufferSize) zugewiesen. Da die bufferSize jedoch immer eine Potenz von 2 ist (bei der Puffererstellung erzwungenes Limit), wird als aktueller Vorgang ProducerCallId & (bufferSize - 1 verwendet )). Es ist dann frei, das Ereignis in diesem Slot zu ändern.

(Der eigentliche Algorithmus ist etwas komplizierter und umfasst zu Optimierungszwecken das Zwischenspeichern der aktuellen Verbraucher-ID in einer separaten Atomreferenz.)

Wenn das Ereignis geändert wurde, wird die Änderung "veröffentlicht". Beim Veröffentlichen wird der jeweilige Slot im Flag-Array mit dem aktualisierten Flag gefüllt. Der Flag-Wert ist die Nummer der Schleife (ProducerCallId geteilt durch BufferSize (da BufferSize die Potenz 2 ist, ist die eigentliche Operation eine Rechtsverschiebung).

In ähnlicher Weise kann es eine beliebige Anzahl von Verbrauchern geben. Jedes Mal, wenn ein Verbraucher auf den Puffer zugreifen möchte, wird eine consumerCallId generiert (abhängig davon, wie die Verbraucher zum Disruptor hinzugefügt wurden, kann das bei der ID-Generierung verwendete Atom für jeden von ihnen gemeinsam genutzt oder getrennt werden). Diese ConsumerCallId wird dann mit der neuesten producentCallId verglichen, und wenn sie geringer ist, kann der Leser Fortschritte machen.

(Wenn die ProducerCallId sogar der ConsumerCallId entspricht, bedeutet dies, dass der Puffer leer ist und der Consumer zum Warten gezwungen wird. Die Art des Wartens wird durch eine WaitStrategy während der Disruptor-Erstellung definiert.)

Für einzelne Verbraucher (diejenigen mit eigenem ID-Generator) wird als nächstes die Fähigkeit zum Batch-Verbrauch überprüft. Die Slots im Puffer werden in der Reihenfolge von der jeweiligen zur ConsumerCallId (der Index wird auf die gleiche Weise wie für Produzenten bestimmt) bis zu der Slote zur aktuellen ProducerCallId untersucht.

Sie werden in einer Schleife untersucht, indem der im Flag-Array geschriebene Flag-Wert mit einem für die consumerCallId generierten Flag-Wert verglichen wird. Wenn die Flags übereinstimmen, bedeutet dies, dass die Produzenten, die die Slots füllen, ihre Änderungen festgeschrieben haben. Wenn nicht, wird die Schleife unterbrochen und die höchste festgeschriebene Änderungs-ID zurückgegeben. Die Slots von ConsumerCallId bis in changeId empfangen können im Stapel verwendet werden.

Wenn eine Gruppe von Verbrauchern zusammen liest (diejenigen mit gemeinsam genutztem ID-Generator), nimmt jeder nur eine einzelne Anruf-ID entgegen, und nur der Steckplatz für diese einzelne Anruf-ID wird überprüft und zurückgegeben.

Aus diesem Artikel :

Das Disruptormuster ist eine Stapelwarteschlange, die durch ein kreisförmiges Array (dh den Ringpuffer) gesichert ist, das mit vorab zugewiesenen Übertragungsobjekten gefüllt ist und Speicherbarrieren verwendet, um Produzenten und Konsumenten durch Sequenzen zu synchronisieren.

Gedächtnisbarrieren sind schwer zu erklären und Trishas Blog hat meiner Meinung nach mit diesem Beitrag den besten Versuch unternommen: http://mechanitis.blogspot.com/2011/08/dissecting-disruptor-why-its-so-fast. html

Wenn Sie jedoch nicht in die Details auf niedriger Ebene eintauchen möchten, können Sie einfach wissen, dass Speicherbarrieren in Java über das volatileSchlüsselwort oder über das implementiert werden java.util.concurrent.AtomicLong. Die Disruptormustersequenzen sind AtomicLongs und werden zwischen Produzenten und Konsumenten über Speicherbarrieren anstelle von Sperren hin und her kommuniziert.

Ich finde es einfacher, ein Konzept durch Code zu verstehen, daher ist der folgende Code eine einfache Welt von CoralQueue , einer Disruptor-Pattern-Implementierung von CoralBlocks, mit der ich verbunden bin. Im folgenden Code können Sie sehen, wie das Disruptormuster das Batching implementiert und wie der Ringpuffer (dh das kreisförmige Array) eine müllfreie Kommunikation zwischen zwei Threads ermöglicht:

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