Puzzler из concurrency-interest

Замечательную штуку вчера продемонстрировал Hans Boehm в листе рассылки concurrency-interest (любопытно, что тема началась с запроса на легализацию разных трюков с s.m.Unsafe -- дискуссии concurrency-interest частенько очень сильно плутают). Ханс писал про опасность методов типа lock.tryLock() -- в смысле сложности их правильного понимания. Вот, например, код:

//Thread 1:
x = 17;
l.lock();

//Thread 2:
while ( l.tryLock() ) {
    l.unlock();
}
... x ...  // какое значение у x в этой точке?

Чтобы ответить на вопрос, обязан ли x быть равным 17 в конце второго потока надо ответить на вопрос "есть ли здесь data race?"

С одной стороны, с точки зрения "универсального" определения программа считается data race free если для всех sequentally consistent трасс над ней отсутствует конкурентный доступ к данным. В данном случае, кажется, это выполняется -- во всех SC трассах сначала в x пишется 17, и только потом x читается -- поэтому кажется, что код data race free, а значит после цикла во втором потоке x должен быть равен 17.

С другой стороны, JMM определяет data race free несколько иначе -- все конкурирующие доступы к данным должны быть упорядочены через happens-before. Поскольку между потоком 1 и 2 нет никаких ребер HB, то доступы к x на запись и на чтение в разных потоках не упорядочены через HB, и, следовательно, они образуют гонку -- а значит, значение x после цикла во втором потоке не определено.

---

Найти правильную точку зрения сравнительно просто -- достаточно вспомнить, что критические секции (и вообще, и в java в частности) имеют семантику "можно войти, нельзя выйти" (roach motel semantics) -- код может вноситься внутрь секции, но не может выноситься из нее. Это значит, что x=17 и l.lock() в потоке 1 могут быть переупорядочены -- присвоение может быть внесено в критическую секцию. Это сразу дает право сказать, что здесь таки есть гонка, и значение x во втором потоке действительно не определено.

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

А ошибка в рассуждениях в том, что, когда мы придумываем SC трассы над кодом, мы делаем два неявных предположения о том, что такое лок. Первое из них -- что лок неявно содержит некое состояние -- каким потоком он сейчас захвачен. Это состояние меняется вызовами lock()/unlock(), а так же запрашивается, и, возможно, меняется вызовом tryLock(). И это предположение верно -- у лока действительно есть это внутреннее состояние.

Но есть еще и второе предположение -- что изменения этого состояния методами lock()/unlock()/tryLock() -- имеют семантику volatile store/load. И вот это уже не верно. В соответствии со спецификацией, lock()/tryLock() имеют (==обязаны иметь) только семантику acquire, а unlock() -- только семантику release. А это значит, что мы не все трассы рассмотрели.

Путаницы добавляет тот факт, что на реализация блокировок обычно делается через CAS, а CAS на x86 имеет семантику full fence. То есть на x86 для любой реализации монитора (из мне известных) lock() будет иметь семантику full fence просто в силу того, что аппаратная модель памяти соответствующего процессора более слабых барьеров не предусматривает. И код Ханса будет скорее всего работать корректно (==не будет содержать гонок). Но это особенности именно аппаратной модели памяти конкретного процессора. Скажем, на пресловутых Azul-ах есть CAS без семантики барьера, и есть односторонние барьеры. И там вполне можно реализовать (правда, не на уровне java, а разве что на уровне интринсиков JIT-а) монитор, у которого lock() будет иметь честную семантику только acqiure.

Собственно, если я правильно понял Ханса, он и обращает внимание на то, что с введением методов типа tryLock() монитор становится легко принять за этакую неявную volatile переменную -- что не соответствует реальной семантике операций с мониторами.

В продолжение темы: глубже в synchronized vs ReentrantLock

После провокационной статьи Мартина у нас с Алексеем Шипилевым и Артемом состоялась любопытная переписка. Результаты ее более полно описал Артем, в своей статье.

Для меня там была одна важная, но неожиданная деталь. С предыдущей своей попытки разобраться в различиях synchronized и RL блокировок у меня создалось ощущение, что они во многом схожи. То есть synchronized имеет иерархию biased->thin(spin)->fat (строго говоря, скорее thin->(biased|fat) ), а RL имеет иерархию spin->park (с опциональным "выхватыванием" (barging). Это значит (как мне казалось), что оба типа блокировок почти идентичны на уровнях thin-fat. Я видел synchronized lock как (thin+fat) реализованный на уровне JVM + biasing, а ReentrantLock -- как (thin+fat) реализованный на уровне java + barging. При этом (thin+fat) часть у них очень близка (так мне казалось).

Оказывается, что это неправда. Действительно, и у synchronized и у RL есть части thin (CAS-based), и fat (OS-level parking). Но они сильно отличаются по политикам перехода между этими двумя режимами. А именно:

ReentrantLock на каждую попытку захвата монитора сначала пытается захватить монитор быстрым CAS-ом, и, в случае неуспеха этого благого начинания -- паркуется. То есть, по-сути, у RL нет "состояния" thin/fat -- он всегда является их суперпозицией.

synchronized же имеет явное поле (часть markword, вроде бы) "текущий тип блокировки" -- со значениями biased/thin/fat. И у него есть определенные правила перехода между этими типами. В частности, лок стартует в состоянии thin. Если привязка блокировок включена, то, по истечении определенного времени (biasing delay) если лок все время захватывается одним потоком -- он становится biased. Если лок захватывается разными потоками, но остается при этом uncontended (т.е. в каждый момент времени им владеет только один поток) - лок остается в состоянии thin. Если же случается столкновение потоков на блокировке, то лок "надувается" (inflate) -- переходит в состояние fat.

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

Вторая тонкость состоит в том, что "надувшийся" (fat) монитор -- это еще не kernel-space lock. Это свой собственный, JVM-ный, библиотечный лок, который сначала немного спиннится в user-space, и только потом, если спиннингом разрулить contention не удалось -- все-таки идет на поклон к батькеOS thread scheduler-у, прося разрулить ситуевину.

И когда говорят об адаптивном спиннинге в приложении к synchronized мониторам -- имеют в виду именно этот спиннинг внутри "надувшегося" монитора -- последний шанс разрулиться перед прыжком в омут с головойkernel space.

Если посмотреть на эту картину, то получается, что с точки зрения алгоритмов, у RL перед synchronized нет ни одного преимущества -- все, что умеет делать RL умеет и synchronized, но synchronized умеет делать и больше. Но при этом RL проще -- что может оказаться критичным, например, если наш конкретный use case этих сложностей не требует, а попадает на область применимости простейшего uncontended CAS lock-а -- RL может выиграть, например, за счет более простого memory layout-а.

То есть если мы пишем некий компонент, который хотим сделать просто потоковобезопасным (thread-safe) -- то есть мы допускаем его использование в многопоточном окружении, но не рассчитываем на особо большую нагрузку -- то пользуем synchronized и не парим себе мозг. synchronized наиболее адаптивен -- может сам приспособиться к различным профилям нагрузки. Плюс к тому, biased locking + поддержка со стороны JIT-а (выбрасывание блокировок по результатам escape analysis, эскалация блокировок, и прочая белая магия) делает synchronized исключительно эффективным в тех случаях, когда конкретный объект используется в однопоточном сценарии.

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

Вывод я для себя делаю такой -- что теоретические рассуждения о том, в каких ситуациях RL лучше synchronized -- это очень условная вещь. Теоретически ни один априори не лучше. Только эксперименты спасут мировую революцию.

P.S. Алексей -- спасибо за терпеливые разъяснения :)

Снова о biased locking

Очередной виток истории начался с провокационного поста Мартина Томпсона (это автор Disruptor-а, если чё) -- и его продолжения, написанного после того, как Мартину указали на некоторые мелкие недостатки его бенчмарков :)

Во-первых, поднятый Мартином вопрос "есть ли сейчас смысл в привязанных блокировках" породил переписку Дэйва Дайса с Клиффом Кликом. В этой переписке довольно отчетливо высказана интересная мысль -- что привязанные блокировки скорее всего скоро отойдут в прошлое. Почему? Потому что это, по меткому выражению Клиффа "software solution to a hardware problem (expensive CAS)".

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

Сейчас же Клифф хвастается тем, что на Azul-ах uncontended {CAS + full_fence} (обычный CAS у них не имеет семантики барьера) выполняется за 5-6 тиков. На текущих Nehalem-ах локальный CAS требует 10-15 тиков (против мифических >100, с которых все когда-то начиналось) -- и есть шанс, что будет оптимизироваться дальше. В таких условиях Клифф пишет, что в версиях jdk для Azul они уже отказались вовсе от biased locking -- нет смысла, обычный CAS-lock выигрывает у любой реализации biasing. Для версии своего jdk под x86 они все еще используют biased locks (правда, какую-то свою реализацию -- не такую, как у Sun/Oracle JDK) -- но видимо, это вопрос времени.

С другой же стороны, обновленные бенчи Мартина показывают, что пока что, на Nehalem-ах, с использованием стандартного Oracle JDK, biased locks обходят RL/synchronized где-то в 10 раз. Хотя к бенчам Мартина я теперь отношусь как-то с осторожностью :(

Хорошая архитектура...

...позволяет откладывать принятие ключевых решений. Хорошая архитектура максимизирует количество непринятых решений. (с) Боб Мартин

Прочитал сегодня на хабрахабре, и просто поразился -- насколько это соответствует моему собственному ощущению. Огромное спасибо Бобу, и автору статьи за вербализацию.

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

Задачка с stack overflow

Ways to improve performance consistency -- некий Peter Lawrey -- довольно продвинутый чувак, если судить по его блогу -- написал message-passing штукакенцию, отдаленно напоминающую дизруптор -- в том смысле, что используется что-то вроде массива как носитель, и потоки передают друг другу сообщения, извещая друг друга мембарами. Правда, в качестве массива он использует ByteBuffer.allocateDirect(). Смысл этого для меня пока не очень понятен, но такова уж его идея.

Проблема у мужика в том, что результаты бенчмарков у него скачут примерно вдвое-втрое. И он спрашивает сообщество "WTF, gays?".

Причем его более тщательные исследования намекают, что скачки производительности происходят после срабатывания GC. Это как бы намекает, что дело тут в memory layout-е, который может меняться после дефрагментации памяти в ходе сборки мусора.

С одной стороны, ситуация похожа на ту, что я наблюдал в своих экспериментах. С другой -- в его коде пишутся не ссылки, а просто числа (ByteBuffer только их и принимает) -- а для long[] в качестве носителя у меня результаты вполне себе повторяемые, скачков в разы там нет. Это несколько удивляет.

Моя текущая версия происходящего состоит в том, что Питер использует для привязки к ядрам taskset. Но taskset привязывает процесс целиком. Это значит, что никак не фиксируется, как раскиданы по ядрам потоки внутри процесса. GC запускается, скорее всего, в отдельном потоке -- и создает нифиговую пиковую нагрузку -- при этом в момент его запуска собственно бенчмаркирующие потоки приостановлены (stop-the-world). Очень логично, что в этот момент ОС решедулит потоки, так, что ничего не делающие потоки скидываются на одно ядро, а потоку GC, аццки вкалывающему, выделяется ядро в безраздельное пользование. Соответственно, после того, как сборщик мусора успокоился, ОС почему-то не возвращает потоки назад, на отдельные ядра. Тут возникает вопрос "почему", но мне кажется, что это сценарий реалистичный -- обычно такие важные алгоритмы пишутся довольно консервативно, в духе "нет веских причин -- ничего не трогай". Т.е. если нет веских причин, перебрасывать потоки между ядрами шедулер не будет -- противное может привести к тому, что в каких-то ситуациях ему придется постоянно перекидывать потоки, что, понятно, хреново.

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

UPD: Питер отписался по результатам. Мое предположение оказалось неверным -- потоки не решедулятся (UPD2: на самом деле, они как раз решедулятся -- но это не оказывает существенного влияния на производительность). Причина оказалась (по его словам) в том, что буфер и счетчики в определенных случаях оказывались лежащими последовательно в памяти. Поскольку паддинг у его счетчиков сделан только после используемого поля value, то буфер, идущий в памяти перед счетчиками может легко заползти на их строку (я писал про такой сценарий в статье про 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, и разброс в этих результатах может быть гораздо выше -- кроме того на персональной машине гораздо сложнее обеспечить должную изоляцию).

taskset для бенчмарков

Экспериментировал с бенчмарками с привязыванием потоков к ядрам через taskset. И вот не дает мне покоя мысль: ведь если я привяжу процесс, тестирующий какой-то двухпоточный код к двум конкретным ядрам (taskset -c 1,2 mvn exec:java ...) то с одной стороны мои потоки не будут кочевать по другим ядрам. А с другой -- у JVM кроме моих 2 потоков еще дюжина фоновых запущена, не считая GC -- и они ведь все тоже будут за эти два ядра конкурировать. Хотя у меня еще 16 ядер свободны... Это не считая того, что никто не помешает ведь моим двум потокам периодически просто меняться ядрами :)

Похоже, все-таки надо искать способ назначать thread affinity mask изнутри, для конкретных потоков. Где-то в сети я видел код, делающий это через JNA довольно компактно.

UPD:Бросил все, и написал свою собственную утилиту, с преферансом и куртизанками: ThreadAffinity.java. Утилита использует JNA, работает только на linux -- винды у меня сейчас банально нет под рукой, а на MacOS найти нужный системный вызов мне пока не удалось.

Open Source: Retryer

У меня наконец-то дошли руки выложить что-то из своих работ в ОС. Библиотечка, созданная изначально еще во времена работы над DataGuard-ом, потом допиленная уже на нынешней работе -- Retryer.

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

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

1. Задать код, который нужно выполнять "пока не получится" (==он завершится без исключений)
2. Задать стратегию задержки (back off) между отдельными попытками. Есть библиотека простых стратегий (без задержки, фиксированная пауза, линейно растущая пауза, экспоненциально растущая пауза...) + билдер/dsl для их комбинации. Например:
   

   Backoff
               .withExponentialGrowingDelay()
               .startingWithDelay( 1, TimeUnit.SECONDS )
               .maxTryes( 5 )
               .maxDelay( 10, TimeUnit.SECONDS )
               .build()
   

3. Задать "фатальные ошибки" -- т.е. при некотором классе ошибок дальнейшее продолжение попыток смысла не имеет. Например, если сервис выбросил исключение ConnectionTimeout -- имеет смысл попробовать еще разок позже, авось сеть заработает. Но если сервис выкинул IncorrectProtocolVersionException, то дальше пробовать смысла нет -- мы явно ломимся куда-то не туда

Пример использования (запрос по URL-у):

private static final String URL_TO_QUERY = "http://google.com/?q=Retryer";

public static String simpleQuery( final String urlString ) throws Exception {
    final URL url = new URL( urlString );
    final InputStream is = url.openStream();
    try {
        final InputStreamReader r = new InputStreamReader( is, "ISO-8859-1" );
        try {
            return CharStreams.toString( r );
        } finally {
            r.close();
        }
    } finally {
        is.close();
    }
}

....

public static String queryRetryableComplex( final String urlQuery ) throws Exception {
    return new Retryer().doRetryable(
            new IRetryableTask<String, Exception>() {
                public String execute( final int tryNo ) throws Exception {
                    return simpleQuery( urlQuery );
                }

                public boolean isFatalReason( final int tryNo,
                                                                     final Throwable reason ) {
                    return (reason instanceof MalformedURLException);
                }
            },
            Backoff
                        .withExponentialGrowingDelay()
                        .startingWithDelay( 1, TimeUnit.SECONDS )
                        .maxTryes( 5 )
                        .maxDelay( 10, TimeUnit.SECONDS )
                        .build()
    );
}

здесь мы будем пытаться запросить гугл с экспоненциально растущей задержкой, начиная с 1 секунды, но не более 10 секунд, и не более 5 раз подряд. При этом мы не будем повторять запрос если он выбросит MalformedURLException (какой смысл-то?).

С удовольствием послушаю всяческую ругань. Обещаю активно и интересно ругаться в ответ :)

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() -- разъяснения

По совету друга я вознес мольбу Господу о ниспослании мудрости, и Господь послал мне этого. Немного. Зато несколько раз -- с первого раза я не понял. Произошло это посредством листа рассылки concurrency-interest -- ну, нечего придираться, пути господни неисповедимы. Начало треда я в архивах не нашел почему-то, так что могу дать ссылку только на продолжение: вот.

Вкратце:

Что касается спецификации:
lazySet был введен после последней редакции JMM, поэтому в JMM нет описания такого отношения порядка, который он порождает. Т.е. по сути javadoc специфицирующий lazySet можно рассматривать как расширение JMM. Аналогичным свойством обладает weakCompareAndSet -- его ordering constraints тоже не формализуются в текущей редакции JMM. Идет обсуждение того, как включить такие вещи в очередную редакцию JMM, но пока в попутчиках согласья нет, и вряд ли очередная редакция JMM выйдет в ближайшее время. Вообще варианты более "тонких" мембаров, частным случаем которых является lazySet (== Fences.orderWrites + store) планируется внедрять в специальный класс Fences, соответственно все обсуждение подобных вещей обычно сосредоточено вокруг него. И черновики будущей спецификации можно посмотреть в его class-javadoc. Я там пока нифига не понял, но попытки буду продолжать.

Что касается текущего состояния:

Да, то, что написано в lazySet javadoc действительно означает, что все записи, произведенные до lazySet() гарантированно станут видимы до того, как станет видимым запись lazySet(). При этом чтения, идущие до lazySet() в program order вовсе не обязаны реально происходить до него -- они могут быть переупорядочены и отложены, и завершиться уже после завершения lazySet() -- в этом, собственно, отличие от обычного volatile store.

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

На самом деле я здесь не вижу существенной разницы с обычным volatile store -- нигде в JMM не прописано, что volatile store должен завершиться немедленно. JMM задает лишь ограничения на то, что все записи и чтения идущие до vstore в program order действительно будут исполнены и их результаты видимы до того, как будет исполнена и видима vstore.

Что касается текущей реализации:

Хотя формально запись может быть отложена на сколько угодно, фактически все существующие JVM реализуют lazySet() консервативно -- выполняют запись немедленно. То есть запись может быть отложена уже только процессором, задержавшись в его store buffer.

Таким образом, использование lazySet в disruptor, похоже, в целом легально, и ускорение, им создаваемое -- вполне объяснимо.

P.S. В процессе разбирательств выяснилось, что некоторые вещи в JMM я до сих пор понимаю больше интуитивно. Кажется, в ближайшее время у меня будут посты на эту тему

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 вызванный выполнением соответствующих барьеров). Если же мы подбираем размер пачки под текущую нагрузку -- мы амортизируем накладные расходы на передачу данных от ядра к ядру на количество этих сообщений, тем самым мы лучше адаптируемся к изменениям в трафике, сохраняя задержку обработки сообщений на минимальном уровне.

False sharing

Размышляю на досуге про false sharing в его простейшем смысле -- про данные, принадлежащие одной строке кэша. Обычно предлагается с этим бороться введением полей (padding) -- добить объект фиктивными полями типа, скажем, long до обычного размера строки в 64 байта. Проблем с этим методом -- не оберешься. Даже если забыть на минутку про то, насколько такая вещь как padding (тем более -- "в рукопашную") не вписывается в концепцию java как высокоуровневого языка, остается еще масса вопросов:

Переносимость: cache-line имеет разный размер на разных процессорах -- от 32 до 128 байт. Можно, конечно, закладываться на максимальный размер, но тратить 128 байт на каждую конкурентную переменную может оказаться многовато. Дело не в том, что это много само по себе -- память нынче дешевая. Дело в том, что это же априори горячие в кэше данные. Т.е. это не 128 байт из основной памяти, которой десятки гигабайт, это 128 байт из кэша первого уровня -- который все еще весьма невелик. Но даже если и так -- кто гарантирует, что завтра не появится процессор со строкой 256 байт? Если приблизиться к реалиям -- по факту везде, где я встречал padding он сделан из расчета на строку в 64 байта. То есть даже на существующих процессорах со строкой кэша длиннее этой -- производительность резко просядет. А ведь я еще нигде не видел, чтобы размер полей можно было хоть как-то "настраивать" -- да и сделать это довольно непросто.

Надежность: padding нынче делается просто набором неиспользуемых примитивных полей. Еще когда я впервые увидел этот трюк где-то в недрах JDK у меня возник вопрос -- а почему собственно JIT не выкинет эти поля нафиг? Ведь даже моя IDE очевидно мне подчеркивала их как неиспользуемые, уж JIT-то должен был бы это увидеть. И если он на данный момент этого не делает -- что будет, если в очередном обновлении он таки научится? Собственно, похоже это произошло: Martin Thompson, один из авторов disruptor уже жалуется, что используемый ими ранее простой padding в очередном обновлении Java 7 работать перестал. Чтобы таки надурить компилятор им пришлось наследовать PaddedAtomicLong от AtomicLong, и делать padding через 6 штук public volatile long (Кстати, почему их 6 а не 7 -- для меня загадка. Могу только предположить, что Мартин заложился на то, что еще 8 байт уйдет на object header. Надо ли говорить, что закладываться на такие детали внутренней реализации JVM не очень надежно?). Пока что это работает -- непонятно только, насколько долго будет.

Или вот такой вопрос -- ну хорошо, добили мы свой объект полями до размера строки. Но ведь мы не можем (из java) гарантировать, по какому адресу будет расположен наш объект. Нам ведь никто не гарантирует, что в памяти наш объект будет выравнен по границе строк. Реально он может оказаться половиной на одной строке, половиной -- на второй. Да, мы можем быть уверены, что если у нас есть два точно таких же объекта, то их одинаковые поля (в предположении что они всегда идут в одном и том же порядке) точно на одной строке не окажутся. То есть, если взять этот самый PaddedAtomicLong -- покуда мы реально используем только "первый" из составляющих его 7 long-ов , мы можем быть уверены, что false sharing это не про нас. Но если бы мы хотели использовать уже 2 его long-а ("первый" и "второй") -- отсутствие false sharing уже не гарантируется, поскольку "второй" может оказаться уже на следующей строке, и на этой же строке может уместиться начало уже другого PaddedAtomicLong, и наш "второй" будет конфликтовать с его "первым". Поэтому, если мы в самом деле хотим избежать false sharing здесь, нам нужен padding с обеих сторон -- и до "защищаемых" нами полей, и после них, размером заведомо больше половины строки в каждом случае.

Но даже это еще не решает проблемы окончательно, потому что у нас ведь в приложении не только одни PaddedAtomicLong-и существуют. Если мы возьмем, для примера, тот же disruptor -- есть еще и объекты сообщений, которые тоже модифицируются (потенциально) многими потоками. И вполне может так случиться, что на одной строке кэша окажется кусок от MyEvent, и начало от PaddedAtomicLong, как раз где самое важное для нас поле value и лежит. И все усилия Мартина и команды по мегаоптимизации своего dsiruptor-а пойдут прахом моментально и внезапно.

...А самое поганое здесь, на мой взгляд, это то, что сам-то disruptor пишет уже упомянутый Мартин Томпсон, который и знает много, и может себе позволить потратить недели на тонкую оптимизацию и выяснение причин каких-либо тормозов. А вот MyEvent будет создавать обычный прикладной программист, у которого скорее всего ни квалификации такой нет, ни времени на исследования. И вероятность того, что этот программист будет свои MyEvent защищать от false sharing мала.

...Но даже если он и будет это делать -- он ведь еще должен это делать таким же способом, как Мартин. Про что я говорю -- я говорю про то, что если, скажем, наш гипотетический программист Вася будет защищать свои MyEvent от false sharing с помощью двустороннего padding-а размером чуть больше полстроки (как я описывал выше) -- то все равно он не избежит кары небесной, ибо Мартин свои объекты защищает односторонним padding-ом. Легко понять, что в этом случае на одной строке не могут оказаться используемые поля от двух разных MyEvent, точно так же как и от двух разных PaddedAtomicLong, но могут оказаться используемые поля от одного MyEvent и другого PaddedAtomicLong. По-сути получается, что страдает инкапсуляция.

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

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

Идеальный вариант, разумеется, чтобы JIT сам определял, какие объекты испытывают наибольшее давление false sharing, и прямо в рантайме перемещал их в памяти так, чтобы это давление уменьшить. Мне этот вариант не очень нравится просто потому, что непонятно, за что тогда будут платить деньги мне вряд ли это будет реализовано скоро, и еще более вряд ли скоро и качественно.

Более реальный вариант: хочется стандартную аннотацию типа @PreventFalseSharing (хотелось бы для нее какое-нибудь название, больше отражающее идею, нежели реализацию), особым образом интерпретируемую JIT-ом -- он для таких объектов использует отдельный аллокатор, как минимум размещающий их строго по границам строк, и не позволяющий никому более использовать место, остающееся до целого количества строк (хотя это уже и не обязательно -- первой части уже будет достаточно, чтобы такие объекты гарантированно не конфликтовали хотя бы между собой). Сюда же можно добавить и включение для ссылочных полей таких объектов оптимизацию write barrier-а в виде test-and-mark вместо только mark. Ну и так далее -- думаю, там еще много чего можно добавить.

Причем на первое время такую штуку можно сделать вообще отдельно от Oracle -- сделать byte-code preprocessor, который при загрузке классов ищет соответствующую аннотацию, и модифицирует байт-код соответствующих классов, добавляя фиктивные поля. Можно даже сделать автоматическое определение размера строки кэша -- через нативный вызов, например. Потом можно прикрутить обработку к javac -- опять же под worst-case, но хотя бы исходный код не будет загрязнен всякими непонятными полями. И уж на последнем этапе можно перевести обработку на уровень JVM/JIT.

Вот такие вот эротические фантазии меня последнее время посещают.

UPD: По сообщениям из заслуживающих доверия источников, в казематах оракла суровыми инженерами сана, там заточенными, обсуждается аннотация @Contended, реализующая что-то вроде варианта 2. Так что не пройдет и 10 лет, как, возможно, мы узрим свет в окошке...

"История одной оптимизации"

На хабре промелькнула статья Как убрать все управляющие символы из строки -- история одной бурной оптимизации.

Каждый раз жалею, что там еще не зарегистрирован -- полдня бы точно потратил на веселье в комментах :)

По-правде говоря, статья произвела фиговое впечатление. Я не хочу вдаваться в подробности оптимизаций, дело не в этом -- мне не понравилось отсутствие какой-то системы. Т.е. если первые 5 шагов оптимизации еще как-то понятны -- это переделка алгоритма с использованием априорной информации о задаче -- то дальше идет почти в чистом виде попытка переплюнуть JIT. И вот уже эта часть -- просто какая-то каша. Раз за разом все более странный код, все более в C-style, но главное -- нет никакой системы, просто набор слепых попыток "а если вот эдак?".

Две вещи мне особенно понравились. Автор сам задает, что 99.9% строк у него в изменении не нуждаются -- они "правильные". Казалось бы отсюда прямо-таки очевидно следует, что оптимизировать нужно не исправление неправильных строк, а поиск правильных -- т.е. маршрут, по которому проходят строки без единого управляющего символа должен быть самым быстрым и самым коротким. Ан нет -- самое лучшее решение, которое предложил автор выглядит так:

public static char[] buf = new char[1024];
...
int length = s.length();
char[] oldChars = (length < 1024) ? buf : new char[length];
s.getChars(0, length, oldChars, 0);
int newLen = 0;
for (int j = 0; j < length; j++) {
    char ch = oldChars[j];
    if (ch >= ' ') {
        oldChars[newLen] = ch;
        newLen++;
    }
}
if (newLen != length)
    s = new String(oldChars, 0, newLen);

Потрясающе -- на этом самом коротком маршруте мы дважды копируем символы исходной строки -- первый раз внутри getChars(), второй руками, внутри цикла -- и все это для того, чтобы в 999 случаях из 1000 обнаружить, что ни первое, ни второе копирование нам не нужно, потому что newLen == length. Мне это кажется немного странным.

Я не поленился прогнать бенчмарки дюжины разных реализаций. Вот лидер:

public class PlainScanLazyReplacer {

    public String replaceControls( final String str ) {
        final int length = str.length();
        for ( int i = 0; i < length; i++ ) {
            final char ch = str.charAt( i );
            if ( Util.isControlChar( ch ) ) {
                return realReplace( str, i );
            }
        }
        return str;
    }

    protected String realReplace( final String s,
                                  final int startWithIndex ) {
        final int length = s.length();
        final StringBuilder sb = new StringBuilder( s );
        int newLen = startWithIndex;
        for(int i=startWithIndex;i < length;i++){
            final char ch = sb.charAt( i );
            if ( Util.isNonControlChar( ch ) ) {
                sb.setCharAt( newLen, ch );
                newLen++;
            }
        }
        sb.setLength( newLen );
        return sb.toString();
    }

Метод realReplace() можно оптимизировать, но смысла в этом нет (я проверял) -- он ведь вызывается 1 раз из 1000. Обратите внимание, что основной цикл -- самый классический прогон по строке с помощью charAt(). Ничего больше -- не нужно.

Версия с reflection:

private static final Field buffField;
    private static final Field offsetField;

    static {

        try {
            buffField = String.class.getDeclaredField( "value" );
            buffField.setAccessible( true );
            offsetField = String.class.getDeclaredField( "offset" );
            offsetField.setAccessible( true );
        } catch ( NoSuchFieldException e ) {
            throw new RuntimeException( e );
        }

    }

    public String replaceControls( final String s ) {
        try {
            final char[] buff = ( char[] ) buffField.get( s );
            final int offset = ( Integer ) offsetField.get( s );
            final int length = s.length();

            for ( int i = 0; i < length; i++ ) {
                final char ch = buff[offset + i];
                if ( Util.isControlChar( ch ) ) {
                    return realReplace( s, i );
                }
            }
            return s;
        } catch ( IllegalAccessException e ) {
            throw new RuntimeException( e );
        }
    }

работает (у меня) за то же время, что и более простая и очевидна версия выше. Лично меня это даже и не удивляет -- потому что charAt() будет почти наверняка встроен JIT-ом, и получится тот же самый проход по массиву -- только уже безо всякой магии reflection.

Мораль сей басни: чтобы быть умнее JIT-a надо в самом деле быть очень, очень умным :)

Вторая вещь еще интереснее. "Важно не то, почему код работает быстро, важно то, почему он не может работать быстрее -- во что мы здесь уперлись?" (Алексей Шипилев). Во что мы упираемся в коде из листинга 1? Мне пришла в голову такая вот идея: когда я генерирую входные данные для бенчмарка, обычно я создаю строки каждую по-отдельности -- со своим внутренним массивом char[] value. А что будет, если я солью все строки в одну большую строку, а потом "нарежу" ее на отдельные строки -- так, что все строки будут разделять один-единственный массив символов (и, более того, их данные будут идти в этом массиве по порядку)? По-идее, это должно давать идеальный шаблон доступа к памяти. Если мы упираемся в какие-то вычисления (в скорость CPU) -- изменение шаблона доступа к памяти не должно особо влиять на производительность. А вот если мы упираемся в память -- должно.

Результат бенчмарка: cache-friendly код выполняется на треть быстрее обычного. Это, конечно, не 100% доказательство, но уже серьезный повод подумать, имеет ли смысл дальше пытаться оптимизировать код разными трюками типа изменения порядка обхода массива, если мы, похоже, упираемся в скорость доступа к памяти?

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, но что-то очень близкое к тому. Все тяжелые вещи -- типа анализа, без которого все равно не обойтись -- можно вынести на отдельные машины, получающие данные с входного/выходного конвейера -- без малейшей задержки центральной части.

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