Jak działa wzór zakłócający LMAX?

Próbuję zrozumieć wzór zakłócający . Obejrzałem wideo InfoQ i próbowałem przeczytać ich artykuł. Rozumiem, że w grę wchodzi bufor pierścieniowy, który jest inicjowany jako wyjątkowo duża tablica, aby wykorzystać lokalizację pamięci podręcznej i wyeliminować przydział nowej pamięci.

Wygląda na to, że istnieje jedna lub więcej liczb całkowitych atomowych, które śledzą pozycje. Wydaje się, że każde „zdarzenie” ma unikalny identyfikator, a jego pozycja w pierścieniu jest ustalana poprzez znalezienie jego modułu w stosunku do wielkości pierścienia itp. Itp.

Niestety nie mam intuicyjnego wyczucia, jak to działa. Zrobiłem wiele aplikacji handlowych i studiowałem model aktora , spojrzałem na SEDA itp.

W swojej prezentacji wspomnieli, że ten wzorzec działa w zasadzie na routerach; jednak nie znalazłem też żadnego dobrego opisu działania routerów.

Czy są jakieś dobre wskazówki do lepszego wyjaśnienia?

Projekt Google Code odwołuje się do dokumentu technicznego na temat implementacji bufora pierścieniowego, jednak jest to trochę suche, akademickie i trudne dla kogoś, kto chce dowiedzieć się, jak to działa. Istnieje jednak kilka postów na blogu, które zaczęły wyjaśniać elementy wewnętrzne w bardziej czytelny sposób. Istnieje objaśnienie bufora pierścieniowego, który jest rdzeniem wzoru zakłócającego, opis barier konsumenckich (część związana z odczytem z zakłócającego) oraz pewne informacje na temat obsługi wielu producentów .

Najprostszy opis Disruptora to: Jest to sposób przesyłania wiadomości między wątkami w najbardziej efektywny możliwy sposób. Może być stosowany jako alternatywa dla kolejki, ale ma też wiele funkcji z SEDA i aktorami.

W porównaniu do kolejek:

Disruptor umożliwia przekazywanie wiadomości do innych wątków, w razie potrzeby budzenie jej (podobnie jak w przypadku BlockingQueue). Istnieją jednak 3 wyraźne różnice.

1. Użytkownik Disruptor określa sposób przechowywania wiadomości, rozszerzając klasę Entry i udostępniając fabrykę do wstępnego przydzielenia. Pozwala to na ponowne użycie pamięci (kopiowanie) lub Wpis może zawierać odwołanie do innego obiektu.
2. Umieszczanie wiadomości w Disruptorze jest procesem 2-fazowym, najpierw zajęte jest miejsce w buforze pierścieniowym, co zapewnia użytkownikowi wpis, który można wypełnić odpowiednimi danymi. Następnie wpis musi zostać zatwierdzony, to podejście 2-fazowe jest konieczne, aby umożliwić elastyczne wykorzystanie pamięci wspomniane powyżej. To zatwierdzenie sprawia, że ​​komunikat jest widoczny dla wątków konsumenckich.
3. Obowiązkiem konsumenta jest monitorowanie komunikatów odebranych z bufora pierścieniowego. Przeniesienie tej odpowiedzialności poza sam bufor pierścieniowy pomogło zmniejszyć ilość rywalizacji o zapis, ponieważ każdy wątek utrzymuje swój własny licznik.

W porównaniu do aktorów

Model Actor jest bliżej Disruptor niż większość innych modeli programowania, szczególnie jeśli używasz dostarczonych klas BatchConsumer / BatchHandler. Klasy te ukrywają wszystkie zawiłości związane z utrzymywaniem zużytych numerów sekwencji i zapewniają zestaw prostych wywołań zwrotnych w przypadku wystąpienia ważnych zdarzeń. Istnieje jednak kilka subtelnych różnic.

1. Disruptor używa modelu konsumenckiego 1 wątek - 1, w którym aktorzy używają modelu N: M, tzn. Możesz mieć tyle aktorów, ile chcesz, i będą one rozmieszczone w określonej liczbie wątków (zazwyczaj 1 na rdzeń).
2. Interfejs BatchHandler zapewnia dodatkowe (i bardzo ważne) wywołanie zwrotne onEndOfBatch(). Pozwala to powolnym odbiorcom, np. Tym, którzy robią operacje we / wy, grupować zdarzenia w celu zwiększenia przepustowości. Możliwe jest wykonywanie wsadowe w innych środowiskach Actor, jednak ponieważ prawie wszystkie inne środowiska nie zapewniają wywołania zwrotnego na końcu wsadu, należy użyć limitu czasu w celu ustalenia końca wsadu, co skutkuje słabym opóźnieniem.

W porównaniu do SEDA

LMAX zbudował wzór Disruptor, aby zastąpić podejście oparte na SEDA.

1. Główną poprawą, jaką zapewnił w porównaniu z SEDA, była możliwość równoległej pracy. Aby to zrobić, Disruptor obsługuje wielokrotne przesyłanie tych samych wiadomości (w tej samej kolejności) do wielu konsumentów. Pozwala to uniknąć konieczności etapów rozwidlenia w rurociągu.
2. Pozwalamy również konsumentom czekać na wyniki innych konsumentów bez konieczności stawiania między nimi kolejnego etapu kolejkowania. Konsument może po prostu obserwować numer kolejny konsumenta, od którego jest zależny. Pozwala to uniknąć konieczności łączenia etapów w przygotowaniu.

W porównaniu z barierami pamięci

Innym sposobem myślenia o tym jest zorganizowana, uporządkowana bariera pamięci. Tam, gdzie bariera producenta tworzy barierę zapisu, a bariera konsumenta jest barierą odczytu.

Najpierw chcielibyśmy zrozumieć oferowany model programowania.

Jest jeden lub więcej pisarzy. Jest jeden lub więcej czytelników. Istnieje wiersz wpisów, całkowicie uporządkowanych od starego do nowego (na zdjęciu od lewej do prawej). Pisarze mogą dodawać nowe wpisy po prawej stronie. Każdy czytelnik czyta wpisy sekwencyjnie od lewej do prawej. Oczywiście czytelnicy nie mogą czytać poprzednich pisarzy.

Nie ma koncepcji usuwania wpisu. Używam „czytnika” zamiast „konsumenta”, aby uniknąć spożywania obrazu wpisów. Rozumiemy jednak, że wpisy po lewej stronie ostatniego czytnika stają się bezużyteczne.

Zasadniczo czytelnicy mogą czytać jednocześnie i niezależnie. Możemy jednak deklarować zależności między czytelnikami. Zależności czytników mogą być dowolnymi wykresami acyklicznymi. Jeśli czytnik B zależy od czytnika A, czytelnik B nie może odczytać poprzedniego czytnika A.

Zależność od czytnika powstaje, ponieważ czytnik A może opatrywać adnotacje wpisem, a czytnik B zależy od tej adnotacji. Na przykład A wykonuje pewne obliczenia dla wpisu i zapisuje wynik w polu awe wpisie. A następnie przejdź dalej, a teraz B może odczytać wpis i azapamiętaną wartość A. Jeśli czytnik C nie zależy od A, C nie powinien próbować czytać a.

To rzeczywiście interesujący model programowania. Niezależnie od wydajności, sam model może przynieść korzyści wielu aplikacjom.

Oczywiście głównym celem LMAX jest wydajność. Wykorzystuje wstępnie przydzielony pierścień wpisów. Pierścień jest wystarczająco duży, ale jest ograniczony, aby system nie został obciążony poza pojemność projektową. Jeśli pierścień jest pełny, autor (autorzy) zaczekają, aż najwolniejsi czytelnicy przejdą i zrobią miejsce.

Obiekty wejściowe są wstępnie przydzielane i są wieczne, aby obniżyć koszty wywozu śmieci. Nie wstawiamy nowych obiektów wejściowych ani nie usuwamy starych obiektów wejściowych, zamiast tego pisarz prosi o istniejący wpis, zapełniając jego pola i powiadamiając czytelników. To pozorne działanie 2-fazowe jest po prostu działaniem atomowym

setNewEntry(EntryPopulator);

interface EntryPopulator{ void populate(Entry existingEntry); }

Wstępne przydzielanie wpisów oznacza także, że sąsiadujące wpisy (bardzo prawdopodobne) są zlokalizowane w sąsiednich komórkach pamięci, a ponieważ czytniki czytają wpisy sekwencyjnie, ważne jest wykorzystanie pamięci podręcznej procesora.

I wiele wysiłków, aby uniknąć blokady, CAS, a nawet bariery pamięci (np. Użyj nieulotnej zmiennej sekwencji, jeśli jest tylko jeden zapis)

Dla twórców czytelników: różni czytelnicy z adnotacjami powinni pisać w różnych polach, aby uniknąć rywalizacji o zapis. (Właściwie powinni pisać do różnych linii pamięci podręcznej). Czytelnik z adnotacjami nie powinien dotykać niczego, co mogą przeczytać inni czytelnicy niezależni. Dlatego mówię, że ci czytelnicy opisują wpisy, zamiast modyfikować wpisy.

Martin Fowler napisał artykuł o LMAX i strukturze zakłócającej, The LMAX Architecture , który może to wyjaśnić bardziej szczegółowo.

Właściwie poświęciłem czas na zbadanie rzeczywistego źródła, z czystej ciekawości, a pomysł, który za tym stoi, jest dość prosty. Najnowsza wersja w momencie pisania tego postu to 3.2.1.

Istnieje bufor przechowujący wstępnie przydzielone zdarzenia, które będą przechowywać dane do odczytania przez konsumentów.

Bufor jest wspierany przez tablicę flag (tablica liczb całkowitych) o jego długości, która opisuje dostępność miejsc na bufory (więcej szczegółów w dalszej części). Dostęp do tablicy jest podobny do java # AtomicIntegerArray, więc dla celów tego wyjaśnienia równie dobrze możesz założyć, że jest ona jedna.

Może być dowolna liczba producentów. Gdy producent chce zapisać do bufora, generowana jest długa liczba (jak podczas wywoływania AtomicLong # getAndIncrement, Disruptor faktycznie używa własnej implementacji, ale działa w ten sam sposób). Nazwijmy to generowane długo nazwą producenta CallCd. W podobny sposób generowany jest konsumentCallId, gdy konsument KONIEC odczytuje boks z bufora. Dostęp do najnowszego adresu konsumenta jest możliwy.

(Jeśli jest wielu konsumentów, wybierane jest połączenie o najniższym identyfikatorze).

Te identyfikatory są następnie porównywane, a jeśli różnica między nimi jest mniejsza niż po stronie bufora, producent może pisać.

(Jeśli wartość parametru ProducCallId jest większa niż najnowszy rozmiar ConsumerCallId + bufferSize, oznacza to, że bufor jest pełny, a producent jest zmuszony czekać na magistrali, aż miejsce stanie się dostępne).

Producentowi zostaje następnie przypisane miejsce w buforze na podstawie jego callId (którym jest prducerCallId modulo bufferSize, ale ponieważ rozmiar bufferSize jest zawsze potęgą 2 (limit wymuszony przy tworzeniu bufora), zastosowana operacja aktuall to producentCallId & (bufferSize - 1 )). Następnie można modyfikować wydarzenie w tym gnieździe.

(Rzeczywisty algorytm jest nieco bardziej skomplikowany i wymaga buforowania ostatniego konsumenta w osobnym odwołaniu atomowym w celu optymalizacji).

Gdy wydarzenie zostało zmodyfikowane, zmiana jest „publikowana”. Podczas publikowania odpowiedni slot w tablicy flag jest wypełniony zaktualizowaną flagą. Wartością flagi jest numer pętli (producentCallId podzielony przez bufferSize (ponownie, ponieważ bufferSize to potęga 2, faktyczna operacja to przesunięcie w prawo).

W podobny sposób może być dowolna liczba konsumentów. Za każdym razem, gdy konsument chce uzyskać dostęp do bufora, generowany jest identyfikator konsumenta Callall (w zależności od tego, w jaki sposób konsumenci zostali dodani do elementu zakłócającego, atom używane do generowania identyfikatorów może być współdzielone lub oddzielne dla każdego z nich). Ten ConsumerCallId jest następnie porównywany z najnowszym producentemCallId, a jeśli jest mniejszy z tych dwóch, czytelnik może robić postępy.

(Podobnie, jeśli producentCallId jest nawet w stosunku do konsumentaCallId, oznacza to, że bufor jest empety i konsument jest zmuszony czekać. Sposób oczekiwania jest definiowany przez WaitStrategy podczas tworzenia zakłócacza).

W przypadku indywidualnych konsumentów (tych z własnym generatorem identyfikatorów) następną sprawą jest możliwość konsumpcji partii. Miejsca w buforze są sprawdzane w kolejności od tej odpowiadającej konsumentowi CallCd (indeks jest ustalany w taki sam sposób jak dla producentów), do tej odpowiadającej najnowszemu producentowi CallCall.

Są one badane w pętli poprzez porównanie wartości flagi zapisanej w tablicy flag, z wartością flagi wygenerowaną dla ConsumerCallId. Jeśli flagi się zgadzają, oznacza to, że producenci wypełniający automaty dokonali zmian. Jeśli nie, pętla jest przerywana i zwracana jest najwyższa zatwierdzona zmiana Id. Automaty od ConsumerCallId do otrzymanych w changeId można wykorzystać wsadowo.

Jeśli grupa konsumentów będzie czytać razem (ci ze wspólnym generatorem identyfikatora), każdy z nich przyjmuje tylko jedno callId, a tylko miejsce dla tego pojedynczego callId jest sprawdzane i zwracane.

Z tego artykułu :

Wzorzec zakłócający jest kolejką wsadową wspieraną przez kolistą macierz (tj. Bufor pierścieniowy) wypełnioną wstępnie przydzielonymi obiektami przesyłania, które wykorzystują bariery pamięci do synchronizacji producentów i konsumentów poprzez sekwencje.

Bariery pamięci są dość trudne do wyjaśnienia, a blog Trishy podjął najlepszą próbę, moim zdaniem, z tym postem: http://mechanitis.blogspot.com/2011/08/dissecting-disruptor-why-its-so-fast. HTML

Ale jeśli nie chcesz zagłębiać się w szczegóły niskiego poziomu, możesz po prostu wiedzieć, że bariery pamięci w Javie są implementowane za pomocą volatilesłowa kluczowego lub poprzez java.util.concurrent.AtomicLong. Sekwencje wzorców zakłócających są takie AtomicLongi są przekazywane w tę iz powrotem między producentami i konsumentami za pośrednictwem barier pamięci zamiast blokad.

Łatwiej jest mi zrozumieć koncepcję za pomocą kodu, więc poniższy kod jest prostym helloworldem od CoralQueue , który jest implementacją wzoru zakłócającego wykonaną przez CoralBlocks, z którym jestem związany. W poniższym kodzie widać, w jaki sposób wzorzec przerywacza implementuje wsadowanie oraz w jaki sposób bufor pierścieniowy (tj. Tablica kołowa) pozwala na bezśmieszną komunikację między dwoma wątkami:

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