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

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

Highly Scalable Java: 3

Интереснейшая еще идея была в презентации Клиффа на JavaOne 2008 -- высококонкурентная lock-free очередь, best-effort FIFO, для использования в архитектуре producer-consumer.

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

while( !CAS( items[producerLocalIndex], null, newItem ) ){

  producerLocalIndex = ( producerLocalIndex + 1 ) % items.length;

}

Т.е. просто пытаемся положить элемент в массив, предполагая что очередная ячейка пуста. Если мы не угадываем (или кто-то уже успел перед нами) -- просто продвигаемся к следующей ячейке.

Берем мы элементы из очереди аналогично:

do{
  consumerLocalIndex = ( consumerLocalIndex + 1 ) % items.length;
  item = items[consumerLocalIndex];
} while( !CAS( items[consumerLocalIndex], item, null ) );
return item;

(consumerLocalIndex изначально каждый потребитель тоже выбирает случайно).

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

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

А еще мне очень понравилось как именно Клифф пришел к этой идее. Логика рассуждений (по крайней мере, как она изложена в презентации) была такой: нам нужна очередь, в которую могут писать/читать много агентов параллельно. Логичная идея -- взять несколько очередей (в идеале -- сколько писателей/производителей) -- массив очередей, и при записи/чтении перебирать очереди. Теперь возьмем число агентов очень большим -- 1000. Во что превратится наш "массив очередей"? Будет очень много очень маленьких (коротких) очередей. Возьмем совсем крайний случай -- очередь может быть либо пустой, либо содержать один элемент, и таких очередей -- дофига. Получаем как раз простой массив, где в ячейке либо null (пусто), либо что-то лежит.

Эта манера рассуждения мне очень напомнила ТРИЗ. Есть там очень похожий прием обострения технического противоречия, доведения до крайности -- "оператор РВС" (размер-время-стоимость).

Что мне в этой идее не очень нравится -- неочевидно, как здесь обеспечить локальность. Т.е. если поток-производитель и поток-потребитель сидят физически на одном ядре, то хотелось бы, чтобы они с большей вероятностью передавали данные друг другу, и только если такой возможности нет -- использовали бы данные других ядер. Так, например, реализованы очереди задач в Fork/Join у Daug Lea. Как реализовать такую штуку в этой модели мне пока неочевидно. Еще более важным это становится если архитектура памяти не SMP, как у AzulBox, а NUMA, что у действительно многоядерных систем встречается гораздо чаще -- тогда, мне кажется, все будет вообще плохо, потому что массив будет скорее всего выделен в локальной памяти инициализирующего потока, а использоваться будет всеми потоками.

Highly Scalable Java: 2

Давно собирался разобраться в устройстве highly scalable java от Cliff Click -- наконец дошли руки.

Итак, Клифф Клик, Главный Инженер (Distinguished Engineer) в компании Azul System. Компания специализируется на готовых решениях для запуска java-приложений. Наисовременейший Azul Box имеет на борту 768 ядер -- специально заточенных под исполнение JVM. Разумеется, JVM у них тоже своя, специальная -- хотя, очевидно, разработана на базе Сановской. Такое количество ядер, конечно, загрузить непросто -- большинство популярных даже высококонкурентных алгоритмов настолько не масштабируются. Клифф предлагает Новый Подход к разработке таких алгоритмов -- на примере своей реализации NonBlockingHashMap (ну и, до кучи, NonBlockingHashSet, и прочих производных).

Подход вкратце описан в презентации на JavaOne 2008, чуть более подробно -- и зубодробительно -- в исходном коде самой библиотеки, которую легко найти в сети. Реализация, которую предлагает Клифф -- lock-free, то есть на каждый шаг кто-нибудь из задействованных потоков прогрессирует.

Идея вкратце: данные храним в простом массиве. Определяем для каждого слота граф состояний, в которых он может находиться в процессе работы -- конечный автомат для каждого слота. Перемещения по ребрам графа состояний выполняются атомарными CAS-ами. CAS либо удался -- тогда мы сменили состояние, либо провалился -- значит, он удался у какого-то другого потока -- отсюда мы получаем общий прогресс и lock freedom. Ключевой момент здесь -- что все, необходимое для определения/изменения текущего состояния слота находится в нем самом -- т.е. все переходы получаются локальными.

Важно: даже традиционно глобальные операции -- такие как resize/rehash/compaction -- здесь организованы так, что распадаются на локальные операции с каждым слотом в отдельности. Соответственно, даже они почти идеально параллелятся между всеми потоками, оперирующими над таблицей.

Собственно, на мой взгляд, это самая интересная часть. В самом деле, реализовать идеально параллелящийся только-на-чтение HashMap совершенно тривиально -- обычный j.u.HashMap вполне сойдет при условии корректной начальной публикации ссылки (safe publishing). Реализовать обновление значений для уже существующих ключей тоже несложно -- один CAS и дело с концом. Удаление в стиле mark removed реализуется так же, как обновление. Все эти операции требуют только локального CAS-а над конкретной ячейкой массива -- ничего сложного.

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

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

Что делают при этом конкурирующие операции?

Операции чтения: если видят обычный (непомеченный) слот в исходном массиве -- его и возвращают, поскольку это все еще актуальное значение слота (и это fast path, реализующийся чаще всего -- очень быстрый). Если видят слот "в процессе копирования" -- значит таблица в процессе ресайза, и нужно во-первых, постараться вписаться -- скопировать значение слота в новый массив: CAS(newArray[index], null, oldArray[index]) (почему это обязательно -- потому что тот поток, что пометил старый слот как "в процессе копирования" мог еще не успеть реально его скопировать -- и тогда это за него сделаем мы). Дальше -- независимо от успеха предыдущей операции -- мы просто читаем актуальное значение слота из нового массива.

Операции обновления/удаления (пометки): опять же, если слот в основном массиве еще не помечен -- можно свободно работать с ним, поскольку его еще не скопировали. Если же пометка есть -- мы, так же, как и при чтении, впрягаемся и сами копируем текущее значение -- и потом меняем его на свое новое.

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

Выглядит это очень круто, хотя в реальном коде все значительно сложнее и запутаннее. Например, Клифф очень аккуратно (экономно) расставляет мембары -- а в джаве нет явных мембаров, поэтому в нужных местах ему приходится читать/писать волатильные переменные, иногда совсем ненужные по смыслу кода -- только ради memory ordering effects. Далее -- он не использует j.u.c.AtomicXXX, вместо этого он активно использует s.m.Unsafe, что не добавляет коду краткости. Зачем? Из-за одной тонкости, о которой я недавно писал -- AtomicXXX.CAS() по контракту обязан делать full fence -- хотя здесь он Клиффу часто излишен. А вот относительно s.m.Unsafe.compareAndSwapObject() контракт ничего такого не требует, хотя сановская реализация его по факту делает. И здесь, видимо, Клифф использует финт ушами -- реализация JVM на Azul Box мембара там не делает -- и поэтому на азулах его таблица будет работать еще быстрее.

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

Всю свою жизнь я писал статические классы-утилиты таким образом:

public final class Helpers{
  private Helpers(){
  }
  public static String veryHelpfullMethod(){
  ....
  }
}

Недавно заметил у коллеги в коде такой вариант:

public final class Helpers{
  private Helpers(){
    throw new AssertionError("Not for instantiation!");
  }
}

Взял себе на вооружение. И правда, с нынешним reflection-ом закрытый конструктор уже не гарантирует, что класс невозможно будет создать. А вот так -- final-класс, с единственным конструктором и выбросом исключения -- я уже не могу придумать способа (кроме byte-code-modification, конечно -- но против лома нет приема).

AtomicXXX.lazySet(...)

Еще один любопытный метод в атомиках -- AtomicXXX.lazySet(T value). Присваивает значение переменной "когда-нибудь" (eventually). Спроектирован для end-of-life cleanup в неблокирующих структурах данных, когда надо очистить значение переменной (например, записать туда null), и больше с этой переменной ничего делаться не будет, поэтому в принципе все равно, когда именно значение будет реально записано (и увидено другими потоками), но важно, что когда-нибудь это-таки произойдет (чтобы GC гарантированно мог собрать "отвязанный" объект). По-сути получается, что это non-volatile write в волатильную переменную.

О сколько нам открытий чудных
Готовит javadoc-ов дух...

1024 cores

Интересный блог о параллельном программировании -- 1024 cores

weakCompareAndSet

В классах AtomicXXX, наряду с известным методом compareAndSet (CAS) есть еще довольно странный метод weakCompareAndSet (WCAS). Не так давно я наткнулся на довольно длинную дискуссию в concurrency-interest. Хотя сама дискуссия довольно уже старая (2006г) но для меня содержала много нового, чем хочется поделиться.

Итак, чем отличается weakCAS от обычного CAS?

1. Может откатиться (ничего не изменить и вернуть false) даже если значение expected совпадает с текущим значением в ячейке памяти (другими словами -- может откатиться "без причины")
2. Не создает (=не обязан создавать) ребра happens-before

Теперь чуть подробнее:

С первым пунктом все довольно просто -- обычный CAS откатывается только в случае, значение в ячейке памяти не совпадает с ожидаемым. Если CAS вернул false -- вы можете быть уверены, что actualValue!=expectedValue. В случае с weakCAS такой уверенности нет -- значения могут совпадать, а weakCAS все равно откатится. Разумеется, можно сделать цикл с проверкой, но есть тонкости -- часто доказательство lock-free для алгоритма базируется на том, что CAS либо удался (и тогда текущий поток "продвинулся"), либо нет (и это означает, что "продвинулся" какой-то другой поток) -- в итоге всегда получается общий прогресс. С weakCAS такое рассуждение уже не прокатит -- формально говоря, он может откатываться сколько пожелает.

Второй пункт важнее и интереснее. Обычный CAS имеет семантику volatile read+write. В частности, это означает, что между двумя CAS-ами одной переменной всегда существует ребро happens-before. Именно это позволяет использовать его, например, для создания спин-локов. weakCAS такой семантики не имеет, никаких ребер не создает, и сделать на его основе спин-лок не получится.

Откуда же есть пошел weakCAS? Ну, если верить concurrency-interest, мотивация была в том, что не на всех моделях процессоров есть инструкция вида CAS. На некоторых есть альтернативный вариант -- load-linked/store conditional. CAS и LL/SC эквивалентны в том смысле, что любой алгоритм, который можно реализовать через CAS можно реализовать и через LL/SC. Но при этом LL/SC откатывается если значение по указанному адресу изменялось с момента чтения (т.е. в случае сценария ABA CAS пройдет, а LL/SC откатится). Есть еще варианты -- например, какие-то реализации LL/SC могут откатываться если изменилась любая ячейка памяти, лежащая на той же cache line, что и нужная. Чтобы на этих процессорах эмулировать честный CAS потребуется цикл с перепроверкой. Но в некоторых случаях "честности" и не нужно, и тогда можно использовать слабый CAS без дополнительных оберток.

Вторая тонкость -- на мой взгляд, более важная -- про happens before. Честный CAS, как я уже писал, состоит, собственно, из двух частей: атомарной проверки-и-присвоения, и двустороннего мембара (full fence). И если первая часть как-то очевидна (само название операции об этом и говорит), то вторая -- это некоторое дополнительная фича именно java-api для AtomicXXX. Не во всех процессорах инструкции CAS обязательно выполняют full fence, и, более того, далеко не во всех алгоритмах, требующих атомарной операции compare-and-set, обязательно требуется и full fence в каждом случае. Между тем, full fence это операция, потенциально, с большой латентностью (особенно, если ядер достаточно много). Поэтому, особенно для многоядерных систем, неплохо бы иметь возможность делать CAS без full fence, когда это необходимо. weakCAS именно такую возможность и предоставляет.

Почему тогда weakCAS малоизвестен? Ну просто потому, что в самых распространенных интеловских и AMD-шных процессорах инструкции "CAS без full fence" просто нет -- там есть только LOCK CMPXCHG, который делает точно то, что требуется спецификацией AtomicXXX.compareAndSet(...). Поэтому на этих машинах weakCAS == CAS.

Вторая причина состоит в том, что довольно сложно решить, где именно может быть применим CAS без fence. В concurrent-interest на эту тему пишут, что "если вы не Daug Lea, и у вас нет крайне веских причин оптимизировать какой-то CAS в вашем коде -- забудьте про weakCAS".

Когда же можно подумать об его использовании? Первый, очевидный, случай -- это счетчики. В частности, в дискуссии concurrent-interest выступал Cliff Click, рассказывая, что ему, для Azul box с 700-800 ядрами, требовались performance counter-ы, которые бы как можно меньше тормозили (на Azul, как я понимаю, есть более богатый выбор вариантов инструкции CAS -- и с барьерами, и без них). В самом деле, если все, что вам нужно -- это атомарный инкремент некоторой переменной -- довольно глупо в пару к нему покупать еще и full fence. Особенно если у вас 800 ядер, частенько выполняющих этот инкремент. Более общо: если вам не нужна синхронизация, а нужна только атомарная операция типа "считать-изменить-записать" -- можно использовать weakCAS, и радоваться от мысли, что, если когда-нибудь ваш код будет выполняться на Azul-ах (какая честь для вашего кода!) -- он будет работать быстрее :)

Второй случай, менее очевидный и более сложный -- когда у вас в алгоритме есть несколько последовательных CAS-ов, и вы понимаете, что "ключевым" (linearization point) из них является только какой-то один -- тогда именно его можно оставить честным, а предыдущие -- weakCAS. Мне это напоминает технику synchronization piggybacking.

Вообще, после таких обсуждений я начинаю понимать Daug Lea, который хочет добавить в джаву класс Fences с методами типа orderReads(..), orderWrites(), orderAccesses(), reachabilityFence(..)