Disruptor 3.0

ТАСС уполномочен заявить, что скоро выходит третья версия Disruptor-a. Michael Barker, который его сейчас развивает, дразнится — обещает прирост в производительности почти вдвое.

Если честно, мне сложно представить, чем еще можно ускорить простейший single publisher single subscriber случай. Но одно предположение у меня есть. Когда я гонял бенчмарки перед JavaOne, я пробовал заменять AtomicLongFieldUpdater.lazySet() на Unsafe.putOrderedLong(). Это то же самое по смыслу, но отсутствуют различные проверки доступа и type-safety. Разница была довольно существенной. Правда, не вдвое, процентов на 20-25. Но, поскольку больше возможностей я не вижу, то я ставлю на то, что в третьей версии Disruptor появится магический класс sun.misc.Unsafe.

Хотя будет гораздо интереснее, если я ошибусь, и будет что-то новенькое :)

Cache coherency #3: false sharing

Я долго-долго тупил, но наконец-то решился продолжить серию про кэш-когерентность. Сами базовые протоколы кэш-когерентности мы разобрали — теперь можно посмотреть на всякие любопытные вещи, которые происходят за кулисами, пока мы на сцене делаем что-нибудь совсем обычное.

False sharing

...Или ложное разделение. По-правде, не очень понятно, почему именно ложное — оно самое что ни на есть настоящее, только неожиданное. Что такое true sharing? — Это когда у нас есть ячейка памяти, используемая одновременно более чем одним потоком. В предыдущих статьях я уже рассказывал, что конкурентный доступ на чтение протоколы когерентности разруливают без особого труда, а вот конкурирующие чтение и запись, и особенно запись и запись заставляют их попотеть — здесь все понятно. А false sharing это когда ячейки памяти разные, но физически попадают на один и тот же cache line. С точки зрения программиста — никакого разделения данных нет, никакого конкурирующего доступа тоже. А с точки зрения подсистемы памяти каждый cache line — это одна ячейка, и операции доступа к разным частям одной строки кэша из разных ядер конкурируют друг с другом за эту ячейку-строку. И неожиданно оказывается, что обычная запись в какую-то локальную (казалось бы) переменную стоит в несколько раз дороже, чем должна была бы:

int threadId = 0, 1, 2, ... THREADS_COUNT;
int distance = 1..8; 
long[] area = new long[THREADS_COUNT*distance];

for(int i=0...Inf){
   area[threadId*distance] = i;
}

(Готовый к исполнению код здесь). Если увеличивать distance, то можно наблюдать, как время на выполнение цикла уменьшается, вплоть до distance=8 — для интелловских архитектур, где cache line 64 байта. (Строго говоря, нужно еще убедиться, что все потоки сели на разные физические ядра, но как правило, в случае незагруженной машины системный планировщик их сам так и раскидает)

Вообще, про false sharing не писал уже только ленивый. Я лично эту тему мусолил аж 4 раза, даже тэг для этого специальный завел. Поэтому я здесь хочу остановиться на другом вопросе, который мне как-то задали в комментариях: если все так легко может стать так плохо, почему это редко когда заметно? (Есть, конечно, вариант, что не замечают потому что и не смотрят — но это недостаточно сексуальный вариант)

У false sharing есть два ощутимых эффекта: увеличение в разы времени выполнения конкретной операции записи (memory latency), и увеличение нагрузки на шину. Чтобы заметить увеличение нагрузки на шину нужно либо специально ее мониторить, либо уже использовать шину близко к режиму насыщения. Но интелловский InterConnect имеет пропускную способность что-то вроде 20Гб/с в каждую сторону — а это 20 миллионов килобайтных сообщений в секунду (в нереальном случае 100% КПД). Не то, чтобы это было недостижимо круто, но не каждая птица долетит до середины Днепракаждое приложение подходит хотя бы близко к тому, чтобы насытить шину.

А что насчет memory latency? Ну, проблема медленной памяти перед разработчиками процессоров стоит уже лет 15, за это время было придумано масса трюков, позволяющих ее как-то обойти. Часть из них — те, что про введение иерархии кэшей — в случае с false sharing не помогут, потому что они сами и являются причиной. Но на уровне процессора остается еще пара тузов в рукаве, связанных с асинхронным выполнением операций с памятью. Например, современные процессоры содержат блок префетчинга, который может предсказать доступ к памяти до того, как исполнение дойдет до соответствующей инструкции, и отправить запрос на упреждающее чтение нужного блока в кэш. Другой пример: инструкции записи тоже обычно не ждут, пока запись фактически будет выполнена — команда на запись просто сохраняется в т.н. store buffer-е, и исполнение идет дальше, а команды из буфера спускаются контроллеру памяти в асинхронном режиме.

В итоге получается, что процессор старается не ждать, пока контроллер памяти тупитвертится как белка в колесе, пытаясь разрулить конфликтующие запросы. Он, насколько это возможно, будет обрабатывать другие инструкции, выполнение которых не зависит от результата "застрявшей" операции с памятью. Поэтому, если кроме конфликтующих доступов к памяти ваш код делает еще достаточно много каких-то локальных (неконфликтующих) операций, то процессор сумеет (возможно сумеет) "замаскировать" среди них неожиданное удорожание доступа к какой-то конкретной ячейке памяти.

И отсюда можно попробовать порасуждать на тему, в каких случаях false sharing имеет наибольшую вероятность стать заметным.

Во-первых, когда нам важно именно время выполнения конкретной (небольшой) операции (latency, response time), а не пропускная способность в целом (throughput). Если наша операция содержит всего-то пару сотен инструкций, и несколько обращений к памяти, то заметное удорожание даже одного из этих обращений все-таки будет заметно. Процессору просто не за чем будет спрятать замедлившуюся операцию. Да, он может параллельно начать выполнять инструкции хоть из другого потока (через hyper threading), тем самым удерживая на уровне общую пропускную способность — но время выполнения конкретной операции от этого не уменьшится. Мы можем получить такую картину: система по-прежнему выполняет N операций в секунду, только время одной операции выросло с 1мкс до 2мкс. Поскольку далеко не каждого волнует время выполнение конкретной операции (особенно, когда масштаб этого времени — микросекунды), то на этой стадии false sharing часто остается просто незамеченным.

Дальше идут ситуации, когда операции с false sharing составляют заметную долю всех операций (по крайней мере, на критическом маршруте). Это как раз пример выше. Почему false sharing там заметен? Потому что там, кроме самой записи, только обвязка цикла, которая едва тянет на пару тактов. И одна-единственная запись, которая contended (счетчик цикла скорее всего будет сидеть в регистре). Даже если запись идет в L1 (~5ns) — она здесь будет доминирующей, и никакие трюки этого не замаскируют, потому что маскировать нечем. И если запись станет в разы дороже — это будет просто бросаться в глаза.

Ну и последний очевидный вариант — это когда у вас шина памяти уже близка к насыщению, и тут еще добавляется false sharing, и все становится грустно.

Т.е. получается, что в первую очередь false sharing беспокоит тех, кто разрабатывает latency-critical системы, и уж во-вторую — тех, кто разрабатывает очень high throughput, где-то близко к пиковой производительности железа.

P.S. Есть у меня еще такая любопытна идея: кажется, что чем мягче аппаратная модель памяти, тем больше у процессора возможностей "прятать" memory latency. Т.е. и прятать false sharing у него тоже больше возможностей. Однако сравнивать разные процессоры сложновато — они обычно отличаются слишком много чем, не только моделью памяти (да и где бы их все взять для тестов-то). Но можно ужесточить аппаратную модель памяти программно — расставляя барьеры памяти. Если взять тот пример, выше, и вместо обычной записи в ячейку массива делать volatile store (придется заменить массив на AtomicIntegerArray, или дергать Unsafe.putIntVolatile) — то сопутствующий lock xadd скорее всего приведет к принудительному сбросу store buffer-а. А значит не будет возможности отложить запись на "попозже", ее придется делать синхронно. Кажется, что это сделать false sharing еще более заметным.

Еще интереснее будет проверить это для Disruptor-а. На Java One меня спрашивали, почему padding в моих экспериментах никак себя не проявил — вот я сейчас думаю, что причина может быть как раз в том, что я экспериментировал с padding после того, как заменил volatile store для курсоров на lazySet (что эффект от оптимизаций некоммутативен — это не новость). Будет интересно, как появится возможность, проверить, что получится если сделать наоборот: вернуть volatile store, и поиграться с padding.

P.P.S. Вообще-то, эта статья планировалась как продолжение цикла про cache coherency. И про false sharing задумывалось как вступление и разминка. Но я что-то увлекся, и false sharing вылез на место главного героя, не оставив места запланированным атомикам. Ну я и подумал — пусть остается как есть, а атомики получат свою главную роль чуть позже :)

Код бенчмарков

Отдельные люди на JavaOne после доклада просили выложить код бенчмарков. Вот, код здесь.

Ногами прошу не бить :)

...расследования

Я уже писал, что мне наконец-таки пришла в голову мысль, почему у Disruptor в бенчмарках такой большой статистический разброс. Еще точнее -- это не статистический гауссовский шум, а совершенно отчетливо различные режимы работы -- вот только не было понятно, откуда там внутри, где все просто, берутся еще какие-то режимы. Так вот, разгадка этого явления мне явилась в 5 утра в прошлую среду, во время подготовки слайдов к докладу. В честь такого счастья мне показалось честным рассказать про разгадку сначала на докладе, а уж потом здесь :)

В общем, одна картинка -- стоит тысячи слов. А уж знали бы вы, чего мне стоило выдрать эту картинку из своей презентации! ГуглоДокументы так просто контент не отдают:)

Идея простая. Поскольку бенчмарки у меня без нагрузки, то скорость перемещения курсоров издателя/подписчика определяется, в основном, внутренней механикой. А она у издателя и подписчика почти идентична. Поэтому двигаться они будут примерно с одиноковой скоростью. А значит в установившимся режиме они будут находиться на каком-то более-менее фиксированном растоянии друг от друга. Ну, разумеется, как-то колебаясь вокруг этого среднего расстояния.

Так вот, если это среднее расстояние таково, что записываемые издателем сообщения находятся на другом кэш-лайне, чем читаемые подписчиком -- то все шоколадно, контроллер памяти работает в приемлемом режиме. А вот если расстояние окажется слишком маленьким -- издатель и подписчик будут работать в пределах одного кэш-лайна. А значит -- контроллер памяти будет играть в пинг-понг: издатель будет для записи требовать кэшлайн в в своем кэше в состоянии M, а читатель -- в своем, в состоянии S. Да, это типичный пример false sharing.

Какое расстояние будет в установившемся режиме -- это, как я понимаю, в значительной степени случайность. Как там в переходном режиме система поколбасится -- так и будет. Спонтанное нарушение симметрии, одним словом :) Вот и получается несколько режимов функционирования с разной пропускной способностью -- в зависимости от случайного gap-а между курсорами, сложившегося в переходном процессе.

Отсюда же понятно, почему отключение бэтчинга для подписчика ускоряет систему в ненагруженном режиме -- бэтчинг это оптимизация, слегка ускоряющая подписчика. А значит он будет с большей вероятностью догонять издателя, и постоянно тыкаться в его спину. А это, как мы уже поняли, не эффективно.

Интересная штука -- когда курсоры идут плотно друг к другу пропускная способность падает. А вот время ответа (latency) -- скорее всего тоже будет падать. Я его не измерял, но кажется очевидным, что чем меньше расстояние между первым и последним курсором в цепочке -- тем быстрее конкретное сообщение будет полностью обработано. А это значит, что бэтчинг здесь это такой размен throughput<->latency. Можно чуть поднять throughput, но ценой увеличения времени обработки одного сообщения. Можно чуть уменьшить время ответа -- но ценой падения и пропускной способности (==менее эффективного использования шины interconnect-а).


И смотрите, как любопытно выходит (дальше мои довольно умозрительные рассуждения): почти наверняка замедление работы в subscriber-batching будет проявляться только в ненагруженном режиме. В нагруженном режиме скорость движения курсоров определяется не их механикой, а нагрузкой -- курсоры двигаются настолько быстро, насколько обработчик успевает делать свою работу. А это значит, что система будет самоподстраивающейся: если нагрузки немного, и обработчики успевают все перемолоть быстро -- они догонят издателя, и будут идти за ним ноздря в ноздрю, уменьшая время ответа. Как только нагрузка вырастает, обработчики начинают не справляться с работой так быстро -- и отстают от издателя. А это сразу дает им чуть-чуть продохнуть за счет более эффективного бэтчинга (больше сообщений за раз) и за счет большего КПД использования interconnect-а. Все это, разумеется, в некоторых, небольших, пределах -- серьезное изменение нагрузки так не скомпенсируешь. Можно даже примерно прикинуть, насколько небольших: худший режим с бэтчингом у нас 15 000 пакетов/мс (~70 нс/пакет), лучший без бэтчинга 35 000 (~30нс/пакет). Разница в 40 нс -- это и есть оценка того, сколько может нам сэкономить сам Disruptor за счет автоматического перехода между режимами.

UPD: Как мне тонко намекнули в комментариях, я забыл важную часть -- гипотеза-то ценна не сама по себе, а тем, что она имеет экспериментальные подтверждения! А именно: если причина ухудшения пропускной способности при включении бэтчинга и нестабильности работы -- действительно такая, как я предположил, то должна быть корреляция между средним расстоянием между курсорами (а схеме это обозначено как gap), и пропускной способностью в конкретном эксперименте. Это можно проверить, и я это проверил. Результат -- вот он (по X -- gap, по Y -- throughput):


По-моему, все очень наглядно. Скачок производительности происходит как раз в области gap=2.5-3.0 Это отлично коррелирует с тем фактом, что размер сообщений у меня примерно 24 байта, т.е. их на 64-байтный кэшлайн умещается как раз чуть меньше 3 штук.

UPD#2: По многочисленным просьбам трудящихся числом 2 -- бенчмарки с пэддингом сообщений. Без batching 19 789 (+/- 483), с batching 20 488 (+/- 1 277).

JavaOne Moscow 2012 -- слайды с презентации

Отчитал вчера "Расчленяя Disruptor" на JavaOne. Слайды здесь. Сразу предупреждаю: слайды делались не по принципу "замена докладчику", а по принципу "лучший копирайт -- это когда без автора не разберешься".

Как разгребу текущие дела -- напишу пару постов по следам. Надо же, наконец, рассказать, почему в бенчмарках Disruptor такая большая дисперсия...

Скандалы, интриги...

Дошли руки посмотреть на бенчмарки самого Disruptor-а -- у него их есть немного. Сколько я помню, они сравнивали Disruptor vs ArrayBlockingQueue. Но в последних версиях в бенчмарках используются уже LinkedBlockingQueue -- якобы, они быстрее.

Интрига в том, что в их бенчмарках это действительно так -- LBQ стабильно лучше ABQ. Но в моих бенчмарках ситуация ровно обратная -- ABQ стабильно лучше. И я не могу найти причин, почему это так.


UPD: Оказывается, что LBQ быстрее на моем ноуте (Core i7 2GHz), а вот ABQ быстрее на сервере (Xeon E5620 2.4GHz). Подозрение мое падает даже не на то, что это разные модели, а больше на то, что сервер, кажется, ccNUMA, а ноут SMP. Хотя внятно объяснить, как это связано я пока не могу

...и вновь продолжается бой

Я тут в процессе подготовки к докладу -- прорабатываю тему Disruptor-а. И вот любопытные вещи обнаруживаются. Вот оказывается, что чтобы скакнуть почти вдвое от производительности ABQ нет необходимости реализовывать сразу хитрую очередь из Art of MP. Достаточно убрать из ABQ Condition-ы. Да, вполне можно оставить ReentrantLock, просто вместо ожидания на Condition notEmpty/notFull сделать spin-wait:

private final T[] items;
    /** items index for next take, poll or remove */
    private int takeIndex = 0;
    /** items index for next put, offer, or add. */
    private int putIndex = 0;
    /** Number of items in the queue */
    private int count = 0;

    private final ReentrantLock lock = new ReentrantLock();

    @SuppressWarnings( "unchecked" )
    public ABQConditionFreeQueue( final int size ) {
        this.items = ( T[] ) new Object[size];
    }

    private int inc( int i ) {
        return ( ++i == items.length ) ? 0 : i;
    }

    @Override
    public void enqueue( final T item ) throws InterruptedException {
        for ( int i = 0; ; i++ ) {
            if ( count < items.length
                    && tryAcquireLock() ) {
                try {
                    if ( count < items.length ) {
                        items[putIndex] = item;
                        putIndex = inc( putIndex );
                        ++count;
                        return;
                    }
                } finally {
                    releaseLock();
                }
            }
            backoff();
        }
    }

    @Override
    public T dequeue() throws InterruptedException {
        for ( int i = 0; ; i++ ) {
            if ( count > 0 && tryAcquireLock() ) {
                try {
                    if ( count > 0 ) {
                        final T item = items[takeIndex];
                        items[takeIndex] = null;
                        takeIndex = inc( takeIndex );
                        --count;
                        return item;
                    }
                } finally {
                    releaseLock();
                }
            }
            backoff();
        }
    }

    private void backoff() throws InterruptedException {
        Thread.yield();
    }

    private void releaseLock() {
        lock.unlock();
    }

    private boolean tryAcquireLock() throws InterruptedException {
        return lock.tryLock();
    }

Большая часть кода взята из ABQ, если что. Причем без backoff() эта штука обходит ABQ вдвое, а с задержкой через Thread.yield() -- почти вчетверо. Простота залог успеха. А я в прошлый заход сразу полез в дебри CAS-free.

Disruptor #3.2: Эксперименты с производительностью -- update

Последняя серия экспериментов закончилась на том, что между самой оптимизированной реализацией очереди и Disruptor-ом все еще оставалось порядка 10% разницы (27 против 30*10^6 сообщений/сек на моем сервере). По косвенным данным, оставшиеся 10% можно было списать на записи ссылок в массив -- в очереди это делается при каждом enqueue, Д же заполняет массив только однажды (Тут надо заметить, что в моем бенче этот эффект так заметен, потому что я тестирую холостой ход -- полезная нагрузка, передаваемая в сообщении у меня всего 1 long. В реальной жизни полезная нагрузка будет побольше, так что одна дополнительная операция записи ссылки потеряется за несколькими операциями записи полей сообщения -- инфа 100%). Еще я грешил на false sharing by card marking -- поскольку ячейки массива идут подряд, их dirty flags должны быть тоже рядом, и должны обновляться при каждой перезаписи. Из этого я сделал вывод, что в 1.7 с +UseCondCardMark все будет шоколадно -- но это фиг проверишь, потому что 1.7 ставить я все еще не готов.

По зрелому размышлению я пришел к выводу, что card marking здесь не при делах. У меня же ссылки в массив пишутся только одним потоком -- который выполняет enqueue. Вот если бы я на dequeue очищал ссылку в null (как, собственно, стоило бы делать в production-ready коде) -- тогда другое дело. Но же я выкинул очистку на одной из первых оптимизаций -- так что у меня по любому single writer получается.

А вот как проверить, что сама запись ссылки порождает 10% оверхеда -- мне пришло в голову вчера. Все довольно просто: у меня же очереди используют пул сообщений (чтобы не плодить лишнюю аллокацию). Пул, естественно, имеет тот же размер, что и очередь (зачем ему больше?), поэтому после первого же прохода все элементы во внутреннем буфере очереди фиксированы -- на следующих проходах они перезаписываются поверх себя же. Достаточно вместо set сделать там test-and-set -- и можно будет проверять.

Эксперименты были запущены, и -- бинго! -- наконец-то мы совершенно догнали Д :) 30*10^6 msg/sec. Предположение о том, что во всем (в оставшихся 10%) виновата запись ссылки в буфер таким образом, похоже, оправдывается. Процесс разборки магии Д на компоненты на этом можно считать закрытым.

Единственное, что остается для меня все еще не понятным -- это нестабильность результатов. Типичные результаты запуска выглядят так:

Turn # 1 takes 67058 ms; 7456232 ops/sec
Turn # 2 takes 15854 ms; 31537782 ops/sec
Turn # 3 takes 16058 ms; 31137128 ops/sec
Turn # 4 takes 16201 ms; 30862292 ops/sec
Turn # 5 takes 16111 ms; 31034697 ops/sec
Turn # 6 takes 15794 ms; 31657591 ops/sec
Turn # 7 takes 60603 ms; 8250417 ops/sec

Это совершенно не похоже на нормальное распределение вокруг среднего :) Ощущение, что система может функционировать в одном из 2 режимов -- либо все хорошо, либо случается какое-то гавно. ЖопойИнтуитивно я ощущаю, что это тот же случай, что и у Питера -- где-то тут недостаточно полей (padding), из-за чего при неудачной дефрагментации после GC внезапно возникает false sharing. Но вот где именно я пока понять не могу. Вроде бы все важные переменные у меня обложены полями по самые гланды...

Disruptor #3.1.4.1: Эксперименты с производительностью -- анализ

Что можно сказать по поводу результатов экспериментов? Картина получается довольно непростая. Первый вопрос: какой из результатов Disruptor брать за основной -- при разнице в 3 раза между производительностью на малых и больших размерах буфера. Мое мнение, что за основу стоит брать именно 30 миллионов ops/sec, которые он выдает на большом буфере. 75 миллионов на малом буфере с батчингом кажутся мне артефактом "холостого хода". Другими словами, я почти уверен, что если начать использовать более-менее тяжелые объекты событий, и более-менее тяжелые обработчики, то этот результат радикально изменится. На это намекает и гиганский разброс результатов различных экспериментов, и то, что LMAX в реальном своем продакшене сообщает о размере очереди в 2М, и то, что при увеличении размера буфера результаты с батчингом и без него сходятся к одному значению пропускной способности в "жалкие" 30 миллионов операций.

Но если принять за рабочую гипотезу, что реперная точка Disruptor -- 30 миллионов оп/сек, то получается, что потестировать эффект от батчинга у меня толком не получилось -- разница между Д с батчингом и без него становится артефактом эксперимента на холостом ходу. Это не очень-то и удивительно -- батчинг, предположительно, должен сглаживать неравномерности во входном потоке данных, но ведь у нас-то входной поток искусственный, и никаких заметных неравномерностей не содержит, негде батчингу развернуться. Но дело в том, что батчинг -- это принципиальное отличие Disruptor-а от очередей -- отличие в интерфейсе/контракте, а не в реализации (для того, чтобы допустить пакетную обработку в очереди потребуется изменить ее интерфейс). И именно это принципиальное отличие надежно оценить и не удалось. Пичалька...

Тем не менее, если мы принимаем производительность Д за 30 млн. оп/сек, то самая базовая, "тупая" реализация single enq single deq queue может дать нам примерно треть от этой производительности. Применяя к этой тупой реализации последовательно: 1) пул для уменьшения аллокации объектов 2) оптимизацию spin-loop-ов 3) volatile write -> lazySet 4) остаток от деления -> битовая маска -- мы получаем ускорение чуть больше чем вдвое -- т.е. 2/3 от производительности Д, причем вклад от этих трех оптимизаций сравним.

Строго говоря, эти оптимизации не вполне независимы -- то есть, применив их в другой последовательности мы скорее всего получим другие цифры для эффекта каждой из них. Поэтому оценивать их вклад точнее, чем по порядку величины смысла я не вижу.

Ближе подойти к Д мне не удалось (я не готов ставить пока не особо стабильный jdk1.7 с его -XX:+UseCondCardMark на рабочий сервер), но косвенные эксперименты с заменой ссылок на long показывают, что очень вероятно, что еще 1/3 производительности теряется на false sharing by card marking.

И если это так, то самым "доходным" решением в дизайне Д признается идея применить заранее заполненный буфер. Поскольку это сразу позволило им и избежать аллокации объектов-событий, и нагрузки на GC для их сборки, и избежать false sharing при модификации ссылок. В сумме это дает почти половину разницы в производительности между Д и самой тупой реализацией SESDQueue. Остальные тонкости -- в том числе padding (на который любит напирать Мартин в своем блоге), и кэширование условий в spin-loop-ах, и lazySet (о которых он не упоминает:) ) -- на фоне этого выглядят довольно бледно.

Очень впечатляет рывок, который получается заменой Object[] -> long[] -- более чем втрое подскакивает производительность. Выглядит это так, что запись ссылочных полей стоит весьма недешево, по крайней мере в многопоточном окружении.

Вот такая получилась загогулина. Напоминаю только, что все это про холостой ход -- под нагрузкой результаты могут сильно отличаться, и ключевые факторы в производительности могут стать другими.

Disruptor #3.1.4: Эксперименты с производительностью (продолжение)

Я расчитывал написать продолжение предыдущей статьи уже через пару дней, но эксперименты затянулись...

В предыдущий заход я остановился на том, что оптимизированная single enqueuer single dequeuer очередь на холостом ходу обходит j.u.c.ABQ примерно в 5 раз -- и во столько же раз ее обходит disruptor. Поскольку я не верю в черную магию (тем более -- чужую), приходится ковырять дальше.

Во-первых, по совету мастеров я начал использовать taskset для привязки потоков к ядрам. Потом я еще немного подумал, и решил напрямую звать sched_setaffinity() через JNA для конкретных потоков. Плюс nice -20 чтобы уменьшить шанс вмешательства всяких фоновых процессов -- это несколько уменьшило дрожание и чуть улучшило результаты. Во-вторых я сделал несколько экспериментов на нашем рабочем сервере -- который (Xeon E5620, 2.40GHz)x8, Java: 1.6.0_18, Sun Microsystems Inc, Runtime: 1.6.0_18-b18, VM: 16.0-b13, OS: Linux 2.6.32

Обновленные результаты предыдущих серий такие (указан диапазон для различных значений размера буфера: 1024б-2Мб, проценты -- среднее значение стандартного отклонения):

ABQ:	(2.6-2.8 ± 7%)
Optimized SESDQueue:	(17-19 ± 20%)
Disruptor:	(30-75 ± 30%)

*10⁶ops/sec

Заходим с хвоста

После долгих раздумий я решил зайти с обратной стороны. Основные конкурентные операции в реализации очереди это volatile read/write указателей на голову/хвост очереди. Можно попробовать убрать вообще всю работу с массивом -- размещение/удаление оттуда элементов -- и оставить только механику передвижения курсоров:

public class SESDSequencer {
    //======================================================
    private static final AtomicLongFieldUpdater<SESDSequencer> tailUpdater = AtomicLongFieldUpdater.newUpdater( SESDSequencer.class, "tailCursor" );
    private static final AtomicLongFieldUpdater<SESDSequencer> headUpdater = AtomicLongFieldUpdater.newUpdater( SESDSequencer.class, "headCursor" );
    //======================================================
    private final int length;
    private volatile long headCursor = 0;
    private volatile long p11, p12, p13, p14, p15, p16, p17, p18 = 7;
    private volatile long tailCursor = 0;
    private volatile long p21, p22, p23, p24, p25, p26, p27, p28 = 8;

    private long lastHeadObserved = 0;
    private volatile long p31, p32, p33, p34, p35, p36, p37, p38 = 9;
    private long lastTailObserved = 0;

    public long sumPaddingToPreventOptimisation() {
        return p11 + p12 + p13 + p14 + p15 + p16 + p17 + p18
                + p21 + p22 + p23 + p24 + p25 + p26 + p27 + p28
                + p31 + p32 + p33 + p34 + p35 + p36 + p37 + p38;
    }

    public SESDSequencer( final int length ) {
        checkArgument( length > 0, "length(%s) must be >0", length );
        this.length = length;
        sumPaddingToPreventOptimisation();
    }

    private long nextTail() {
        final long tail = tailCursor;
        waitWhileNotFull( tail );
        return tail + 1;
    }

    private long nextHead() {
        final long head = headCursor;
        waitWhileNotEmpty( head );
        return head + 1;
    }

    private void publishTail( final long newTail ) {
        tailUpdater.lazySet( this, newTail );
    }

    private void publishHead( final long newHead ) {
        headUpdater.lazySet( this, newHead );
    }

    private void waitWhileNotFull( final long tail ) {
        //spin-wait: "while not full"
        final long target = tail - length;
        while ( target == lastHeadObserved ) {
            lastHeadObserved = headCursor;
        }
    }

    private void waitWhileNotEmpty( final long head ) {
        //spin-wait: "while not empty"
        while ( head == lastTailObserved ) {
            lastTailObserved = tailCursor;
        }
    }

    public void moveHead() {
        final long newTail = nextTail();
        publishTail( newTail );
    }

    public void moveTail() {
        final long newHead = nextHead();
        publishHead( newHead );
    }    
}

Эта штука работает в разы быстрее: (142-157± 10%)*10⁶ -- более того, она примерно в два раза быстрее disruptor-а. Ну наконец-то, чудес, похоже, и правда не случается -- мы теперь можем сказать, где собака порыласьчто разница в производительности где-то здесь:

public void enqueue( final T item ) {
        final long newTail = nextTail();
        
        //это те 2 строчки, что отличают 
        //SESDSequencer.moveTail() от SESDQueue.enqueue():
        final int index = index( newTail - 1 );
        elements[index] = item;

        publishTail( newTail );
    }

    public T dequeue() {
        final long newHead = nextHead();

        //а это -- те 2 строчки, что отличают 
        //SESDSequencer.moveHead() от SESDQueue.dequeue():
        final int index = index( newHead - 1 );
        final T item = elements[index];

        publishHead( newHead );
        return item;
    }

Битовые маски

Дальнейшие поиски внутри указанных 4-х строчек приводят к неожиданному результату: замена вычисления индекса через остаток от деления ( sequence % length) на использование битовой маски ( sequence & (length-1) -- соответственно, ограничивая размер буфера степенью двойки) дает около 25% прироста производительности: (19-27± 20%)*10⁶ops/sec.

Кто бы мог подумать? -- а я еще сомневался, стоит ли экономить на такой мелочи как деление. Посыпаю голову пеплом.

GC write barrier

Следующая идея состоит в том, чтобы проверить гипотезу о влиянии GC write barrier-а при записи ссылки. С этой целью я сделал реализацию очереди, которая хранит не ссылки на объекты (LongValueEntry) а просто long -- соответственно, внутри у нее не Object[], а long[]. Результат: (107-114± 10%)*10⁶ops/sec.

Похоже, мы зажали disruptor в вилку, и очень похоже, что предположение о GC write barrier действительно оправдывается.

Деоптимизируем Disruptor

Еще одна идея возникла уже со стороны Disruptor-а -- почему бы не отключить в нем batching? Это несложно: нужно скопировать com.lmax.disruptor.BatchEventProcessor и заменить буквально одну строчку в методе run():

final long availableSequence = sequenceBarrier.waitFor(nextSequence);
    while (nextSequence <= availableSequence)
    {
        event = ringBuffer.get(nextSequence);
        eventHandler.onEvent(event, nextSequence, nextSequence == availableSequence);
        nextSequence++;
    }
    //вот эту строку надо внести в цикл выше ^^
    sequence.set(nextSequence - 1L);

Результат интересный -- на малых размерах очереди (1024-32K) Disruptor без batching сильно тормозится: 28 вместо 70 миллионов ops/sec. На больших же размерах очереди (256К-2М) разница практически отсутствует (особенно учитывая немалую статистическую погрешность примерно в 20%). Можно сформулировать иначе -- Д без батчинга почти не зависит от размера буфера, в то время как с батчингом он сильно (почти в 3 раза) ускоряется на коротких буферах. Пока мне не приходит в голову внятное объяснение этого феномена. Можно заметить, что у Xeon E5620 кэш L1=64K, то есть малые буфера в него полностью влезают, а большие -- полностью не влезают. Этот факт явно что-то символизирует, но вот что именно -- пока не соображу

Выводы

Что у меня пока получается в итоге: 

ABQ:	(2.6-2.8 ± 7%)
SESDQueue(+lazySet, +spinning):	(17-19 ± 20%)
.............+bitmask:	(19-27 ± 20%)
Disruptor (+batching):	(30-75 ± 30%)
Disruptor (-batching):	(26-30 ± 20%)
LongQueue:	(107-114 ± 25%)
Sequencer:	(143-157± 15%)

*10⁶ops/sec

Интересно отметить как зависит производительность от размера буфера. Disruptor (+batching) очень сильно замедляется при увеличении буфера -- примерно в 3 раза при переходе от 1024 до 2Мб. Очереди же замедляются существенно меньше -- на 15-20% всего.

Так же меня все еще не удовлетворяет стабильность результатов -- между различными запусками результаты зачастую отличаются в разы. Привязка к ядрам и повышение приоритета процесса где-то раза в полтора уменьшили этот разброс, но он все равно остается очень большим -- стандартное отклонение для некоторых экспериментов (тут особо выделяется как раз Д) достигает 35%, а амплитуда разброса значений -- 45%. Причем заметно, что повторы внутри одного запуска (одной JVM) имеют значительно меньший разброс. Напрашивается вывод, что это не случайная, а систематическая погрешность, каким-то образом зависящая от момента запуска JVM. Но у меня пока кончились идеи "что еще можно сделать, чтобы уменьшить влияние внешних факторов". Мартин что-то еще писал про перенаправление прерываний -- чтобы они не посылались тем ядрам, на которых сидят бенчмарки...


Продолжение (окончание, надеюсь) -- в следующих сериях


P.S. Нужно сказать, что Алексей Шипилев проводил эти тесты на своем оборудовании, и (на JDK1.7, -XX:+UseCondCardMark ) получил у самой оптимизированной из реализаций очередей (с маской вместо остатка от деления, с lazySet, с большим padding-ом в 128 байт, но без оптимизации spin-wait) результаты практически не отличимые от Disruptor. Я у себя не могу повторить его результаты -- на рабочий сервер ставить экспериментальный пакет OpenJDK1.7 я не готов, а порт для мака на моем ноутбуке дает результаты заметно хуже (но надо помнить, что у меня на маке нет taskset, и разброс в этих результатах может быть гораздо выше -- кроме того на персональной машине гораздо сложнее обеспечить должную изоляцию).

LMAX Disruptor #3.1: эксперименты с производительностью

Разбирательства с lazySet и прочими тонкостями моделей памяти несколько отвлекли меня от продолжения вивисекции disruptor-а. А ведь сам пристальный интерес к lazySet у меня возник как раз после того, как бенчмарки показали, что если заменить его на обычный .set() -- пропускная способность падает почти вдвое. Так что пока суд да дело -- я напишу об этих самых бенчмарках.

Сначала про задумку. Задумка возникла еще во время первого знакомства с disruptor-ом, и укрепилась в ходе дискуссии с Алексеем о его мнимых и реальных достоинствах. Дело в том, что бенчмарки disruptor-а, которые приводят разработчики, сравнивают его с j.u.c.ArrayBlockingQueue. Это достаточно справедливо в том смысле, что ABQ наиболее подходящий готовый компонент из стандартной библиотеки, которым воспользуется обычный разработчик, буде ему понадобится сконструировать конвейерную многопоточную обработку сообщений. Но ABQ это вовсе не самая производительная (ни в плане пропускной способности, ни в плане задержек) реализация конкурентной очереди. Это достаточно простая и надежная реализация, с приемлемой производительностью -- именно поэтому (как я полагаю) она и является частью стандартной библиотеки. И если уж ставить вопрос "disruptor vs queues" -- как его пытаются поставить разработчики, то надо сравнивать disruptor c сопоставимыми по степени оптимизированности реализациями очередей. Которых -- увы -- в стандартной поставке JDK нет. Но это всего лишь означает, что их надо написать :)

Таким образом и возникла идея этого теста. Вместо тестирования "disruptor vs queues" в благоприятных для него условиях попробуем тестировать в неблагоприятных. Возьмем disruptor в простейшей топологии A->B (один поток -- поставщик, один поток -- обработчик-потребитель). Возьмем простейшую реализацию single enqueuer single dequeuer очереди на основе массива (собственно, это будет очень близко к реализации самого disruptor-а -- по-сути, он и есть такая очередь, только на стероидах). Будем постепенно вносить в эту очередь различные оптимизации, или/и будем постепенно "убирать" различные оптимизации из disruptor-а. Поскольку эти две структуры данных весьма близки изначально, то на каком-то этапе они станут почти неразличимы, и производительность должна оказаться равной. То есть на выходе у нас получится последовательность шагов, каждый из которых на сколько-то улучшает очередь/ухудшает disruptor -- так, что они в конце-концов становятся неразличимы. И можно будет смотреть, какие факторы собственно являются в дизайне disruptor-а ключевыми для производительности, а какие -- второстепенными.

Дисклаймер: я провожу тесты без нагрузки. То есть измеряю пропускную способность в сферическом случае в вакууме -- она может коррелировать с производительностью в боевых условиях, а может и не коррелировать. Меня здесь интересуют именно накладные расходы на холостом ходу -- "стоимость" работы внутренней механики. Просто хотя бы потому, что рабочий ход мне не на чем исследовать -- нет конкретного примера.

Single Enqueuer, Single Dequeuer Queue

Итак, откуда начинаем плясать: вот так выглядит single enqueuer single dequeuer array-based queue:

public class SESDQueue<T> {
    private final int length;
    private volatile long headCursor = 0;
    private volatile long tailCursor = 0;

    private final T[] elements;
    /* Elements range: [headCursor, tailCursor)
    *
    * (tailCursor - headCursor) == elements count
    *
    * 0 <= (tailCursor - headCursor) <= length  => state invariant
    *
    * tailCursor - headCursor == length         => queue is full
    * tailCursor - headCursor == 0              => queue is empty
    *
    * (headCursor % length ) is the index of first item in queue
    * (tailCursor % length ) is the index of _cell_ for _next last item_
    */

    public SESDQueue( final int length ) {
        checkArgument( length > 0, "length(%s) must be >0", length );
        this.length = length;
        elements = ( T[] ) new Object[length];
    }

    private long nextTail() {
        final long tail = tailCursor;
        waitWhileNotFull( tail );

        assert ( tail == tailCursor ) : "tail=" + tail + " <> tailCursor=" + tailCursor + " -- seems like multi-threaded enqueue() detected!";
        return tail + 1;
    }

    private long nextHead() {
        final long head = headCursor;
        waitWhileNotEmpty( head );
        assert ( head == headCursor ) : "head=" + head + " <> headCursor=" + headCursor + " -- seems like multi-threaded dequeue() detected!";
        return head + 1;
    }


    private void publishTail( final long newTail ) {
        assert ( newTail >= headCursor ) : "newTail(" + newTail + ") < headCursor(" + headCursor + ")";
        tailCursor = newTail;
    }

    private void publishHead( final long newHead ) {
        assert ( newHead <= tailCursor ) : "newHead(" + newHead + ") > tailCursor(" + tailCursor + ")";
        headCursor = newHead;
    }


    private void waitWhileNotFull( final long tail ) {
        //spin-wait: "while not full"
        while ( tail == headCursor + length ) {
        }
    }

    private void waitWhileNotEmpty( final long head ) {
        //spin-wait: "while not empty"
        while ( head == tailCursor ) {
        }
    }

    public void enqueue( final T item ) {
        final long newTail = nextTail();

        final int index = ( int ) ( ( newTail - 1 ) % length );
        elements[index] = item;

        publishTail( newTail );
    }

    public T dequeue() {
        final long newHead = nextHead();

        final int index = ( int ) ( ( newHead - 1 ) % length );
        final T item = elements[index];
        elements[index] = null;//for GC

        publishHead( newHead );
        return item;
    }

    public int size() {
        return ( int ) ( tailCursor - headCursor );
    }
}

В ней нет ни одного CAS, только volatile load/stores - это ее плюс. Писать/читать ее может только по одному потоку -- это ее минус, в данном случае несущественный, так как disruptor тоже в своем наилучшем варианте single publishing, и принципиально single subsriber.

Схема бенчей

Один поток пишет в "очередь" 300 миллионов записей (для начала -- LongValueEntry), другой поток их из "очереди" вытаскивает. Время выполнения определяется по моменту вытаскивания последней записи из "очереди".

Цикл выполняется 5 раз подряд в одном запуске -- 1-2 раз обычно выбрасывается, чтобы не учитывать разогрев JVM. Запусков обычно 2-3 -- сколько достаточно для более-менее уверенной статистики.

В бенчмарках присутствует так же ArrayBlockingQueue -- в качестве реперной точки.

Код здесь.

Поехали...

Система: MBP, i7 2GHz, MacOS X 10.6.8,
JDK 1.6.0_26-b03-377, "-da -server -Xmx512m -Xms512"

Емкость очереди -- 1024 элемента.

Базовые результаты (ops/sec == количество сообщений, прошедших "очередь" за секунду).

ABQ	SESD	Disruptor
(2.65+/-0.05)e6	(9.0+/-0.3)e6	(65+/-3)e6

Нифиговое такое различие, а? Меня, правда, немного напрягает, что разница в скорости между ABQ и Disruptor составляет 25 раз, вместо 5-6 в их собственных тестах...

Ок, начинаем оптимизацию. Шаг №1: первое, что приходит в голову -- аллокация. Disruptor использует однажды преаллоцированные объекты, обе наши очереди -- создают 300 миллионов временных объектов. Проверим какое влияние это оказывает -- сделаем пул объектов. Задача упрощается тем, что пул имеет заранее известную емкость (==емкости очереди), плюс он не должен быть потоковобезопасным, потому что мы знаем, что у нас только один поток-производитель:

class EventPool<T> implements EventFactory<T> {
        private final EventFactory<T> factory;
        private final Object[] items;

        private int counter = 0;

        private EventPool( final int size,
                           final EventFactory<T> factory ) {
            this.factory = factory;
            this.items = new Object[size];
            for ( int i = 0; i < items.length; i++ ) {
                items[i] = factory.newInstance();
            }
        }

        @Override
        @SuppressWarnings( "unchecked" )
        public T newInstance() {
            try {
                return ( T ) items[counter];
            } finally {
                counter = ( counter + 1 ) % items.length;
            }
        }        
}

ABQ	SESD	Disruptor
(2.70+/-0.04)e6	(10.7+/-0.4)e6	(65+/-3)e6

Различие не то, чтобы особо радикальное, но процентов 15 наша SESD очередь выиграла.

Шаг №2 -- добавляем padding, чтобы избежать false sharing:

private volatile long headCursor = 0;
    
    public volatile long p11, p12, p13, p14, p15, p16, p17, p18 = 7;
    
    private volatile long tailCursor = 0;
    
    public volatile long p21, p22, p23, p24, p25, p26, p27, p28 = 8;

    public long sumPaddingToPreventOptimisation() {
        return p11 + p12 + p13 + p14 + p15 + p16 + p17 + p18
                + p21 + p22 + p23 + p24 + p25 + p26 + p27 + p28;
    }

результат SESDQueue: (10.8+/-0.2)e6 -- какие-то крохи, даже говорить стыдно. Даже как-то странно...


Шаг №3 -- оптимизируем spin-loop-ы. Опять же, подглядывая в код disruptor-а можно заметить такой прием:

private void waitWhileNotEmpty( final long head ) {
        //spin-wait: "while not empty"
        while ( head == tailCursor ) {
        }
    }

Зачем нам здесь каждый раз читать tailCursor (volatile load), когда единожды обнаружив, что он на сколько-то больше head мы можем просто запомнить его текущее значение в локальной для потока переменной, и пересчитывать ее значение только когда head будет снова ему равно:

//тут опять же будет padding...
    private long lastTailObserved = 0;
    //и тут --- тоже...
    private void waitWhileNotEmpty( final long head ) {
        while ( head == lastTailObserved ) {
            lastTailObserved = tailCursor;
        }
    }

Результат: (12.9+/-0.1)e6 (== +20%).

Шаг №3 -- используем lazySet. Опять же подсмотренно в классе Sequence из disruptor -- похоже, что операции обновления курсоров здесь не требуют всей строгости семантики volatile store -- достаточно LoadStore+StoreStore барьеров перед ними -- то есть гарантии, что все, что произошло до обновления курсора в program order будет видимо остальным потокам до того, как станет видимым это обновление. Для этого не нужно volatile store, достаточно lazySet:

Результат: (13.9+/-0.1)e6 (== +10%).

В общем-то, прямолинейные оптимизации закончились -- а мы все еще отстаем от disruptor в 5 раз. Продолжение -- в следующих сериях :)

AtomicXXX.lazySet() strikes back

Наткнулся тут на любопытную вещь -- я всегда был уверен, что в JMM все поведение всегда можно описать с помощью несложного набора правил построения ребер happens-before. Ну, то есть, я не говорю, что применение этих правил всегда просто -- законы Ньютона тоже не очень сложные, а какой мрак и ужас из них получается, скажем, в небесной механике и движении твердого тела -- но сами правила довольно просты, и применимы к любому java-коду. Но похоже это не совсем так. Этот самый AtomicXXX.lazySet() мутит воду.

Простое объяснение -- неволатильная запись в волатильную переменную -- меня только запутывает. До этого момента в джаве, вообще-то, не было волатильной/не волатильной записи в отрыве от типа поля (ну, если не трогать сатанинский Unsafe). То есть если поле волатильное -- то все операции с ним волатильные, поле не волатильное -- все операции не волатильные.

А теперь получается такая штука, что я могу записать значение в переменную не волатильно, а потом считать его волатильно. И что за отношение порядка будет у меня между этими операциями, если они в разных потоках?

Это не праздный интерес. В коде Disruptor-а встречается такая идиома:

AtomicLong sequence = new AtomicLong();
...
//поток 1: издатель
...write some values into shared objects...
sequence.lazySet(seq);

//поток 2: обработчик
while( sequence.get() != seq ){
   //spin wait
}
...read shared objects fields

По смыслу кода понятно, что автор его уверен -- последовательность sequence.lazySet()/get() обеспечивает завершение всех операций записи, идущих до нее в program order, и видимость результатов этих операций из второго потока. Другими словами -- это практически happens-before edge. Но вот с чего он так в этом уверен -- для меня совершенно не очевидно.

Как минимум, документация lazySet() не дает никаких гарантий относительно того когда запись будет реально выполнена. Eventually -- когда-нибудь. То есть возможно, что и вообще никогда. И непонятно, в какой момент запись -- если она завершится -- будет увидена вторым потоком.

А с другой стороны -- я сейчас гоняю бенчмарки, разбирая по частям disruptor и вытаскивая какие компоненты ключевые для его производительности. И вот оказывается, что если вместо lazySet использовать обычный set() -- скорость падает в 2 раза примерно (это без нагрузки -- на холостом ходу). То есть смысл определенно есть -- но вот надежен ли этот способ оптимизации -- мне как-то неочевидно.

Задал вопрос на stackoverflow -- посмотрим, может там понимают

LMAX Disruptor #2: Реализация

Общую идею, которая стоит за дизайном disruptor я описывал в предыдущей статье, теперь хочу разобрать непосредственно реализацию. Итак, из чего построен disruptor?

Компоненты

Event -- объект-сообщение, который пересылается от производителя ко всем обработчикам через disruptor. Обычно событие создается-пересылается-обрабатывается-выкидывается, а здесь же все события создаются заранее, при инициализации RingBuffer (для этого есть интерфейс фабрики событий EventFactory), и живут, как минимум, столько же, сколько и сам буфер, а все, кто работает с RingBuffer только обновляют поля уже существующих событий (и поэтому с точки зрения реализации было бы точнее называть их Entry, и так оно и было в старых версиях -- но теперь авторы стараются быть ближе к пользователям). Ранее события должны были реализовывать AbstractEvent, но последней версии (2.5) было решено от этого отказаться, и теперь в буфер можно помещать совершенно произвольные объекты -- фреймворк больше ничего конкретного от них не хочет.

Sequence По смыслу это что-то вроде volatile long с некоторыми дополнительными оптимизациями -- например, защитой от false sharing и ослабленной семантикой set(). Объекты этого класса используются для нумерации последовательности событий.

Вообще сама концепция последовательности (Sequence) довольно фундаментальна для disruptor. По-сути, смысл ее в том, что все события в рамках конкретного экземпляра disruptor имеют глобальную логическую нумерацию. Логическую в том смысле, что физически-то экземпляры события переиспользуются -- как я уже писал, их сразу создается ровно столько, сколько нужно чтобы заполнить кольцевой буфер. Т.о. один и тот же физический экземпляр события в разные моменты времени может иметь разный логический номер. Нумерация идет с момента старта конкретного экземпляра disruptor-a, и покуда не закончится long (а он закончится весьма нескоро: даже при обработке 50 млн событий в секунду его хватит на сотни лет -- для практических целей можно считать его бесконечным).

Зная размер буфера этот порядковый номер элементарно преобразовать в индекс в массиве. С другой стороны, покуда речь не заходит об отображении номера на конкретный экземпляр события, можно оперировать только самой последовательностью, и это довольно удобно. Любые два события всегда можно упорядочить -- т.е. поток событий не только сериализуем, но эта сериализация еще и задана в явном виде. Логически мы получаем (почти)бесконечную, однозначно упорядоченную последовательность событий. Внутренняя же логика disruptor-а следит за тем, чтобы физически в каждый момент времени было задействовано не более bufferSize событий.

RingBuffer -- сам кольцевой буфер.

Во-первых, это сам массив записей (сообщений) с методами доступа. Размер массива должен быть степенью двойки, потому что авторы очень хотят иметь возможность вычислять остаток от деления через битовую маску. (Лично мне это стремление не очень понятно -- я где-то со времен P2 был уверен, что экономить на таких мелочах уже бессмысленно).

Во-вторых, здесь сосредоточена та самая внутренняя логика, которая позволяет иметь логически бесконечную последовательность событий, аккуратно отображенную на ограниченный буфер -- логика управления последовательностью. Здесь три части:

1. Текущий курсор. Это логический номер последнего опубликованного сообщения. Сообщения с номерами (cursor-bufferSize, cursor] доступны для дальнейшей обработки (это не совсем точно, есть еще выделенные но не опубликованные сообщения, но про это далее)
   
2. Список курсоров всех обработчиков событий. С его помощью мы следим, чтобы не переиспользовать еще кем-то не обработанную ячейку -- а именно, мы не можем опережать последний (минимальный) из этих курсоров более, чем на bufferSize.
   
3. Стратегия ожидания -- WaitStrategy -- каким способом мы будем ожидать, если кончились доступные нам события (либо производитель не успел еще опубликовать новые, либо наоборот, обработчики еще не успели обработать все опубликованное). Доступны BusySpin/Yelding/Sleeping/Blocking реализации.
4. Стратегия выделения номеров -- ClaimStrategy. Это, пожалуй, самая сложная часть, разберу ее далее.

SequenceBarrier -- это объект, инкапсулирующий в себе набор курсоров (Sequence), которые нам не нужно опрежать. Т.е. если наш обработчик зависим от обработчиков №1 и №2, то у курсора нашего обработчика (последовательности сообщений, им обрабатываемой) будет барьер сверху из курсоров обработчиков №1 и №2. И прежде , чем перейти к обработке следующего события (сдвинуть наш курсор на следующее событие) мы должны дождаться, пока курсоры, от которых мы зависим, уйдут вверх. Вот именно эта операция не лезть поперек батьки "убедиться, что наши предшественники ушли вперед" и является основной функциональностью SequenceBarrier-а. Поскольку ему иногда придется подождать, пока предшественники в самом деле уйдут, внутри него так же лежит стратегия ожидания WaitStrategy. Барьер создается RingBuffer-ом, так что его WaitStrategy та же самая, что использует и сам RingBuffer (вообще, WaitStrategy задается однажды при создании RingBuffer-а, и она используется в большинстве случаях, когда кому-то кого-то приходится ждать).


EventHandler -- это, собственно, интерфейс для обработчика событий, и ничего более.

EventProcessor -- это хелпер, который отвечает за пересылку событий из RingBuffer в EventHandler. Он реализует Runnable, и его нужно засовывать либо в Executor, либо явно в Thread.

Как это работает?

Начнем с производителя

Публикация сообщения в буфер disruptor-а происходит в три этапа: сначала мы резервируем (claim) под себя слот (а еще точнее -- номер события), в который собираемся что-то опубликовать, потом записываем в слот данные, и подтверждаем публикацию (publish). Авторы предпочитают говорить о двух этапах, намекая на то, что запись значений в слот -- внутреннее дело производителя. Пример кода с вики:

// Publishers claim events in sequence
final long sequence = ringBuffer.next();
final ValueEvent event = ringBuffer.get(sequence);

event.setValue(1234); // this could be more complex with multiple fields

// make the event available to EventProcessors
ringBuffer.publish(sequence);   

Зачем нужна такая хитрость? Ну, это обратная сторона решения использовать заранее преаллоцированные объекты событий. Т.е. если рассуждать отвлеченно, для публикации события нужно а) выделить место в памяти куда будем класть содержание б) положить содержание в) сгенерировать номер события г) сделать событие доступным для обработчиков. В случае использования, например, очереди (ABQ), выделение памяти и наполнение обычно производятся где-то в стороне (в клиентском коде), а генерация номера события и публикация объединены -- событие доступно с момента помещения его в очередь, и в этот же момент определяется его порядок относительно других событий. Поэтому в случае с очередью эти 4 этапа как бы не заметны -- первые два вообще не имеют отношения к очереди, а последние 2 объединены.

В случае disruptor ситуация иная: выделение места под данные производится через сам RingBuffer (который, в этом сценарии, выступает в роли пула/кэша/аллокатора объектов сообщений), при этом, в силу дизайна самого disruptor-а номер события однозначно определяет объект события (как buffer[sequence % bufferSize]), мы не можем выделить событие не задав его номер -- поэтому выделение и генерация номера в disruptor-е объединены в фазу 1. С другой стороны, чтобы сделать событие доступным (опубликовать его, фаза "г") нам нужно во-первых, подвинуть курсор RingBuffer-а, во-вторых -- нужен мембар, гарантирующий что все записи в поля сообщения были реально выполнены. Оба эти элемента получаются автоматом за счет volatile write при обновлении курсора.

Более подробно: функциональность выделения объектов событий, и их последующей публикации -- это как раз и есть ответственность ClaimStrategy. Реально, конечно, стратегия с объектами событий не работает -- она работает только с их номерами. Стратегия заявок держит внутри себя свой собственный курсор, указывающий какие номера событий уже выделены. При заявке на очередной номер стратегия делает внутри себя incrementAndGet(), проверяет, что объект с данным номером свободен -- т.е. все обработчики с предыдущего круга его уже обработали (если это не так, мы будем ждать -- ClaimStrategy не использует общую WaitStrategy, здесь, например используется spin->yield->sleep(1)) -- и выдает получившийся номер клиенту.

Коммит (публикация) несколько сложнее -- здесь ответственность за разные фазы поделена между ClaimStrategy и самим RingBuffer. А именно: ClaimStrategy отвечает за порядок публикации сообщений -- они должны публиковаться ровно в том порядке, в котором они пронумерованы. Т.е. если сообщение №3 уже готово к публикации, но сообщение №2 еще нет -- мы будем ждать (by spin-and-yield), пока сообщение №2 таки не опубликуется, и только потом сможем перейти к публикации нашего №3. После того, как ClaimStrategy убедилась, что порядок в порядке -- RingBuffer просто обновляет свой курсор (здесь даже CAS не нужен, достаточно volatile write -- потому что мы можем быть уверены в текущем значении курсора by design). Это обновление, по совместительству, обеспечивает мембар, необходимый для публикации изменений в самом объекте сообщения (чуть более строго, по JMM: видимость изменений в полях объекта обеспечивается ребром HB между VW(cursor) при публикации, и VR(cursor) который делает каждый обработчик в ходе своего spin-wait-а). Это разделение ответственности за коммит на две части мне не очень понятно -- по-моему, было бы логичнее чтобы всю ответственность взяла на себя ClaimStrategy

Разумеется, вся эта машинерия сильно упрощается, если издатель только один -- тогда все запросы идут из одного потока, и ClaimStrategy может быть сильно проще. Разработчики это предусмотрели, и существует SingleThreadedStrategy, которая этим преимуществом пользуется

Чисто технически, логика работы с последовательностями (выделение, публикация) вынесена в базовый класс Sequencer, а RingBuffer наследует Sequencer, и добавляет функциональность хранения самих объектов сообщений, и отображение sequence -> event. Сделано это для того, чтобы можно было легко реализовать свою собственную логику хранения сообщений. Например, если сообщения у нас это просто некий ID команды -- мы можем использовать для его хранения long[] без необходимости boxing/unboxing в/из Long

Обработчики

Ок, сообщение мы опубликовали. Что насчет обработчиков?

За пересылку сообщений из RingBuffer к соответствующим EventHandler-ам отвечает реализация интерфейса EventProcessor. Основная рабочая реализация BatchEventProcessor.
BatchEventProcessor держит внутри себя курсор (Sequence) на последнее обработанное сообщение и SequenceBarrier, задающий список курсоров, от которых зависит наш курсор (==список производителей/обработчиков, которых мы должны пропустить вперед нас, таким образом задаются взаимозависимостями между разными обработчиками).

Основной цикл работы BatchEventProcessor-а состоит в том, что он ждет на SequenceBarrier доступности события, следующего за последним обработанным (как я уже писал, SequenceBarrier как раз и позволяет дожидаться, пока событие с таким-то номером будет обработано всеми впереди идущими == SequenceBarrier.waitFor(cursor+1)). Дождавшись этого, он проверяет, какие события нам на самом деле доступны (гранулярность ожидания может оказаться слишком большой, и реально окажется доступен не только cursor+1, но и, скажем, вплоть до cursor+142), и в цикле все их пересылает в EventHandler. После чего -- один раз -- обновляет курсор, сдвигая его на количество обработанных сообщений (это и есть местный batching). Ну и плюс здесь еще всякая вспомогательная машинерия, типа обработки исключений и остановки обработчика.

Эпилог

Это, собственно все. Несмотря на кажущуюся сложность, идейно все довольно-таки просто. По всему коду повторяется один и тот же шаблон: есть курсор на последнее обработанное (в том или ином смысле) сообщение, есть список курсоров, которые ни в коем случае нельзя опережать. Мы сидим в spin-wait-е на списке "необгоняемых курсоров" (возможно, делая back off через yield/sleep(1)/park), и ждем пока минимальный из них не обгонит нас. Как только он это делает -- мы проверяем насколько именно он нас обогнал, и все эти сообщения -- в нашем распоряжении. Благодаря тому, что курсор везде volatile, пара операций "обновление значения курсора -- чтение нового значения курсора в ходе spin-wait" создает ребро HB, а значит все изменения, сделанные потоком, обновившим курсор до этого обновления будут гарантированно видны всем, прочитавшим новое значение курсора после этой самой операции чтения.

Единственное место, где требуется здесь CAS -- это выделение сообщений из пула для случая многих потоков-поставщиков -- но здесь и нельзя ничего сделать, потому что это в чистом виде consensus problem, которая (для потенциально неограниченного числа участников) обычными volatile write/read не решается.

Все остальное -- это оптимизационные тонкости, где какую стратегию ожидания с каким backoff использовать, где какой padding создать.

Интересно, что в disruptor производитель ничем принципиально от обработчика не отличается. В самом деле, обработчики же вполне свободны модифицировать сообщения, так что можно сделать "производителя" в виде заглушки

while(!Thread.currentThread().interrupted()){
  // Publisher claim events in sequence
  final long sequence = ringBuffer.next();
  // make the event available to EventProcessors
  ringBuffer.publish(sequence);   
}

Такой производитель (он будет однопоточным, кстати) большую часть времени будет проводить в ожидании, чтобы не "съесть свой собственный хвост" -- зато все сообщения будут доступны для обработчиков, следующих за ним. Почему так не сделано? Потому что это не позволяет эффективно использовать batching для производителей:

while(!Thread.currentThread().interrupted()){
  final byte[] message = network.receiveEvents(...);
  final int eventsCount = findEventsCountIn(message);
  // Publisher claim events in batch 
  final SequenceBatch sequenceBatch = new SequenceBatch(eventsCount);
  ringBuffer.next(sequenceBatch);
  //parse and copy event content from message to
  for( int i=sequenceBatch.getStart();i<=sequenceBatch.getEnd();i++){
    final Event event = ringBuffer.get(i);
    //copy i-th event content from message to event object
    ...
  }

  // make the events available to EventProcessors
  ringBuffer.publish(sequenceBatch);   
}

То есть размер "пачки" для публикации зависит от текущей ситуации на рынке в сети -- от текущего уровня нагрузки. Если мы реализуем производителя как обработчика -- мы вынуждены публиковать сообщения по одному -- каждый раз неся накладные расходы на volatile read/write (реально, скорее, на cache coherence traffic вызванный выполнением соответствующих барьеров). Если же мы подбираем размер пачки под текущую нагрузку -- мы амортизируем накладные расходы на передачу данных от ядра к ядру на количество этих сообщений, тем самым мы лучше адаптируемся к изменениям в трафике, сохраняя задержку обработки сообщений на минимальном уровне.

LMAX Disruptor #1: Идеи

Я уже упоминал в предыдущем посте о Disruptor -- замечательном фреймворке для организации высокопроизводительного многопоточного конвейера. Поскольку я такие вещи обожаю -- хлебом не корми -- я прочитал все статьи авторов, скачал его исходный код, и медитировал над ним в свободное и рабочее время последние пару недель. Могу сказать, что это и в самом деле очень достойный пример для подражания. Интересные архитектурные решения, достаточно прозрачный код, и весьма нетривиальные хитрости, которые позволяют ему не просто работать, а работать очень быстро. У меня есть большое желание попробовать описать, что показалось мне интересным, важным или просто заслуживающим внимания.

Для начала исходные данные. Сам проект -- open-source, хостится на google-code, вот здесь http://code.google.com/p/disruptor/. На вики проекта есть страничка, где аггрегируются ссылки на все статьи его разработчиков, где проясняются многие моменты его дизайна и использования. Если вы хотите сами поразбираться -- добро пожаловать. Между прочим один из основных разработчиков, и популяризаторов -- Триша -- женского пола А-а-а, крокодил в ванной женщины в программировании!!! :)

Важное замечание: disruptor нынче уже дорос до версии 2.0, причем в ходе взросления система именования основных интерфейсов тоже взрослела -- разработчики пытались подобрать названия, максимально проясняющие суть дела (я тоже обожаю этот naming poker :). Я к тому, что в статьях разной давности некоторые сущности могут называться иначе, чем здесь (я стараюсь использовать нотацию версии 2.0) -- как правило, это не принципиально, но может иногда запутывать.

Я буду стараться следовать логике разработчиков, комментируя места, которые мне не очевидны. Поехали:

Disruptor -- фреймворк для создания многопоточных конвейеров обработки событий. Из схожих вещей -- Акторы, или просто граф (ориентированный) Producers-Consumers. Классический способ организовать такой граф -- запустить код поставщиков-потребителей в различных потоках, и использовать bounded concurrent/thread-safe queue для создания ребер (неклассический вариант -- упоминавшийся мной фреймворк JetLang) . Именно этот подход и тестировали разработчики disruptor прежде, чем начать изобретать свой собственный велосипед. Производительность этого варианта их не устроила. Почему производительность системы очередей может быть неудовлетворительной (точнее -- значительно ниже теоретического предела), и что мы с этим можем сделать?

Общие принципы

Мы априори принимаем, что любая очередь используемая в production должна быть ограниченной емкости (bounded). В самом деле, если мы допускаем неограниченный рост -- мы допускаем, что наше приложение может в любой момент упасть с OutOfMemory. Если этот вариант нас не устраивает -- нужно задавать явные ограничения.

Во-первых, если мы хотим получить максимальную пропускную способность и минимальные задержки в высококонкурентном случае -- нам нужно стараться уйти от блокировок к более мелкозернистым операциям, типа CAS.

Во-вторых, если мы хотим минимальные задержки -- нам нужно минимизировать нагрузку на GC. Как следствие -- очереди основанные на связных списках уже в проигрыше, поскольку они генерируют много мусорных узлов.

В-третьих, желательно, чтобы текущие положения головы и хвоста очереди перемещались по памяти предсказуемым образом. Имеется в виду, что современные процессоры пытаются угадывать паттерн обращений к памяти, которые делает программа, и упреждающе загружать соответствующие данные в кэш. В частности, на данный момент процессоры умеют распознавать паттерн доступа с фиксированным шагом (fixed strides) размером <2Kb. То есть, если наша программа обращается к ячейкам памяти с постоянным шагом, скажем, 128 байт, то через несколько таких обращений процессор начнет понимать, чего мы от него хотим, и заранее запрашивать загрузку очередной такой ячейки (точнее -- строки, ее содержащей) в кэш. Если же наша программа бежит по памяти каким-то более хитрым образом -- процессор уже не осилит его угадать, и мы будем регулярно получать кэш-промахи со всеми вытекающими последствиями для производительности. Мы опять же приходим к выводу, что организация очереди на основе связанных списков менее предпочтительна, чем на основе массива. Это несколько спорные утверждения. Далеко не каждое приложение имеет настолько низкую "фоновую" нагрузку на GC, чтобы дополнительная нагрузка в виде узлов очереди стала заметна -- раз. Можно организовать быстрый пул узлов (тем более раз очередь ограниченного размера -- можно этот пул заранее наполнить) -- два. Можно даже постараться организовать выдачу узлов из пула в некотором определенном порядке, согласованном с их порядком в памяти -- три. С другой стороны все эти обходные пути, конечно, усложняют дело -- трудно судить, насколько они эффективны. Очевидно, имеет смысл рассматривать сначала более простые решения (массив с преаллоцированными ячейками), и только если они не дают нужного результата -- рыть глубже.

Далее -- нам надо стараться минимизировать трафик на межядерной шине interconnect. Для этого нужно стараться избегать параллельной записи в одну ячейку памяти. Более того, в силу существования эффекта false sharing нужно избегать параллельной записи даже просто в близкие ячейки памяти. Кроме того, нужно минимизировать число содаваемых мембаров.
Я еще об этом буду писать подробнее, но предварительно могу сказать, что большая часть хитростей, примененных разработчиками посвящена именно этому пункту -- игре вокруг уменьшения трафика на шине. По-видимому, это один из ключевых моментов в том, как disruptor зарабатывает свою рекордную производительность.

Архитектурные идеи

До сих пор мы рассуждали, как можно сделать максимально быструю конкурентную очередь. Но нам нужна не очередь сама по себе -- нам нужна возможность организовывать графы производителей-обработчиков произвольной топологии, и очереди -- это лишь один из способов их реализации. Посмотрим, можем ли мы получить что-то дополнительное из того, что мы знаем, каким примерно образом мы собираемся использовать проектируемую нами структуру данных?

Можно заметить, что если использовать очереди для соединения узлов графа, то в каждом узле мы вынуждены делать столько операций dequeue/enqueue, сколько выходящих/выходящих ребер у этого узла. Каждая операция enq/deq на конкурентной очереди -- это операция изменения разделяемого состояния -- а это барьеры памяти и трафик на межпроцессорной шине. Однако enc/deq редко когда бывают в самом деле нужны -- обычно очередной обработчик вовсе не "забирает" очередное событие себе, а лишь "берет попользоваться" и вскоре "отдает назад" (иногда -- модифицировав его как-то). И вся топология графа обработчиков состоит в том, что одни обработчики должны отработать гарантированно прежде других -- т.е. должен соблюдаться какой-то порядок вызова обработчиков.

Эти рассуждения приводят к мысли, что конвейер обработки произвольной топологии можно реализовать в виде одного единственного общего буфера событий. Производители и обработчики имеют каждый свой курсор, указывающий индекс текущего для них события в этом буфере. Чтобы гарантировать порядок обработки (==топологию конвейера) достаточно знать, какие производители/обработчики идут впереди тебя, и следить за их курсорами, чтобы не забегать вперед паровоза.

Эта идея не кажется чем-то особенным, пока мы не заметим одну тонкость. Предположим пока, что производитель у нас один-единственный. У нас есть набор курсоров, по одному на каждого участника конвейера. Эти курсоры по сути, и составляют разделяемое состояние конвейера. Любой шаг -- что производителя, что обработчика -- изменяет один из курсоров. Но каждый участник меняет только свой курсор -- чужие курсоры он только читает! То есть у нас получается лучший возможный паттерн работы с разделяемым состоянием -- полное состояние конвейера распадается на несколько частей, и у каждой части один писатель и много читателей.

Если производителей больше одного (что не очень типично -- обычно говна на вентилятор подбрасывают в одну лопату) ситуация, конечно, малость ухудшается -- разрулить несколько производителей, желающих одновременно что-то вписать в буфер без write contention тут не получится, но кое-какие оптимизации и здесь можно сделать -- хотя и без них получается неплохо, ибо производителей (даже если их несколько), чаще всего все равно меньше, чем обработчиков -- а значит мы все равно в плюсе.

Есть у этой схемы еще одна чудесная особенность -- пакетная обработка событий. В самом деле, пусть какой-то обработчик по своим внутренним причинам вдруг взял да ушел в себя на пару миллисекунд. Затем он возвращается, и проверяет курсоры всех производителей, и всех обработчиков, которых он должен пропустить вперед -- и обнаруживает, что самый последний из них убежал от него уже аж на 9000 элементов. Делать нечего, надо догонять -- но ведь нет никакой необходимости делать это элемент за элементом, каждый раз обновляя свой курсор и генерируя трафик на шине -- можно обработать все 9000 одним махом -- ведь гарантированно они все ему доступны -- и обновить курсор лишь однажды! Получается очень удобная штука: автоматическое сглаживание неоднородностей в потоке событий. Если кто-то из участников конвейера вдруг резко рванул вперед, а остальные от него отстали -- они естественным образом ускорятся, потому что обработка событий получается тем дешевле, чем больше их можно обработать за раз, ведь стоимость межпроцессорной коммуникации при обновлении курсора будет амортизироваться на большее число обработанных элементов. От сильных неоднородностей это, конечно, не спасет -- в этом случае затраты на коммуникацию будут каплей в море -- но это уж работа прикладного программиста, составить конвейер из максимально мелкозернистых обработчиков.

Собственно говоря, про идею это все. Вокруг этой аккуратной реализации идеи -- со всеми возможными оптимизациями -- и постоен весь Disruptor. Про эту реализацию -- в следующих сериях, а то и так что-то я сегодня больно дохрена написал, пора и честь знать...

P.S. Чтобы не казалось, что все вообще радужно -- не все, что можно реализовать очередями, можно реализовать эффективно с помощью Disruptor. Первый же попавшийся мне на глаза пример -- автоматическая балансировка нагрузки. Эту штуку элементарно сделать с помощью одной-единственной очереди -- в один конец сколько угодно поставщиков кладут некие задачи, с другого конца N потоков-обработчиков (где N = числу ядер в системе) их разбирают. Нагрузка балансируется по отдельным потокам автоматически -- если кому-то почему-то достаются более легкие задачи, этот кто-то просто быстрее возвращается за новыми. С помощью Disruptor такую штуку не реализуешь -- для каждого обработчика явно, при конструировании конвейера, задано, какие события он получит. Полученные им события никак не будут зависеть от относительной загрузки различных элементов конвейера -- отстающих просто будут ждать. Конечно, никто не помешает создать 10 одинаковых обработчиков, каждый из которых будет обрабатывать лишь одно событие из 10, и пропускать остальные -- но это не будет автоматическим балансированием нагрузки, это будет балансированием числа событий.

LMAX trading

Статья про архитектуру трейдинговой платформы LMAX. Кажется, про нее рассказывали на JavaOne, но я пропустил этот доклад. Читать здесь

Несколько любопытных вещей. Во-первых, они ухитряются обрабатывать 6 миллионов транзакций в секунду на обычных серверных Nehalem-ах, при том, что вся бизнес-логика у них в одном потоке. А, да, все это на java.

Во-вторых, архитектура, с помощью которой они этого добились довольно необычна. Я думаю, тут мне информации еще на несколько постов, по мере того, как я буду разбираться.

Вкратце: во-первых, у них три главных части в приложении -- входная "мультиочередь" (хитрая структура данных, которую они назвали Disruptor, по-сути -- циклическая конкурентная очередь-конвейер, она одна стоит разговора), в которую приходят все заявки, обработчик бизнес-логики (тот самый, однопоточный), который берет заявки из этой очереди, и выходная такая же мультиочередь.

Обработчик бизнес-логики -- это, по-сути, message-loop, однопоточный, написаный максимально оптимизированно. Все данные ("мир") он держит исключительно в оперативке, весь код очень простой и максимально короткий, чтобы JIT его наиболее удачно оптимизировал (JIT-friendly code). Это, в частности, и позволило им получить 6 миллионов транзакций в секунду. Но здесь возникает вопрос надежности (устойчивости к сбоям), и вопрос ввода-вывода -- он, по-любому, медленный, и традиционно в одном потоке не делается.

И тут на сцену выходят как раз мультиочереди-конвейеры. Идея состоит в том, что, прежде чем попасть на вход к обработчику бизнес-логики, сообщение проходит целый конвейер обработчиков. Конвейер этот, по сути, ориентированный граф с определенной топологией. На вход конвейеру сообщение подается чуть ли не в виде массива байт, как оно принято из сети -- причем делать это могут сразу несколько потоков (т.е. они могут слушать сразу несколько сетевых интерфейсов, в частности). Дальше, по мере продвижения по конвейеру, данные могут обрабатываться различными обработчиками.

Большая часть из обработчиков read-only -- и они вполне себе работают параллельно, загружая разные ядра процессора. В частности, какие-то обработчики дублируют данные на вторичные сервера -- реально в продакшене в каждый момент времени работают 3 территориально разнесенных сервера, с полностью идентичным состоянием -- это решает проблему устойчивости к сбоям. Другие обработчики сохраняют данные в персистентное хранилище -- так что состояние системы (которое, напомню, все в оперативке) в любой момент можно восстановить заново накатив весь лог событий (реально раз в сутки делается снапшот, так что накатывать приходится только за последние сутки). Третьи отправляют часть событий для какого-то анализа и последующей генерации отчетов. И так далее.

Часть обработчиков -- read-write, в частности, те, которые собственно парсят и раскодируют сообщение, превращая его из массива байт в джава-объекты, и в конечном счете, в набор команд для процессора бизнес-логики.

Причем команда разработчиков сумела реализовать этот конвейер обработчиков событий очень эффективно, с минимальными накладными расходами, при сохранении высокого уровня конкурентности. Их Disruptor по их бенчмаркам обеспечивает примерно в 4-8 раз большую пропускную способность, чем реализация аналогичного графа с помощью стандартных ArrayBlockingQueue, и на несколько порядков меньшие задержки. Плюс намного меньшую нагрузку на GC.

Что в итоге: в итоге main code path для обработки входящей транзакции включает в себя начальную предобработку (парсинг/проверка/конвертация в команды), которая делается конвейером параллельных обработчиков, работающих на разных ядрах, и передающих друг другу данные с минимальными накладными расходами (через Disruptor), потом отработку команд центральным процессором бизнес-логики -- который максимально короткий, и работает только с оперативкой, и выдачу результата в выходной Disruptor -- что опять же почти даром. Т.е. этот центральный путь -- максимально короткий, и максимально предсказуемый -- на нем нет ничего, что могло бы вызвать непредсказуемые задержки, ни аллокации больших объемов памяти (что приближает full stop-the-world GC), ни блокировок, ни сетевого ввода-вывода в базу данных. Практически, этот центральный путь не вполне real-time, но что-то очень близкое к тому. Все тяжелые вещи -- типа анализа, без которого все равно не обойтись -- можно вынести на отдельные машины, получающие данные с входного/выходного конвейера -- без малейшей задержки центральной части.

Вот такая вот красота. По-моему, просто шедевр инженерного искусства -- всего пара-тройка сравнительно простых и понятных центральных идей, и возможность приклеивать практически все, что угодно -- но "сбоку", не задевая работоспособность и производительность ядра. Ядро можно вылизать, заоптимизировать и затестировать по самое нехочу -- благо, оно сравнительно небольшое.