Left-Right

Пару лет назад ребята из concurrency freaks опубликовали в c-i описание интересного алгоритма управления конкурентным доступом к произвольной структуре данных. Они назвали этот алгоритм Left-Right, и утверждали, что он позволяет реализовать blocking writes, но зато аж wait-free (population oblivios) reads. При этом писатели не блокируют читателей, хотя читатели блокируют писателей (но не блокируют друг друга). Это довольно-таки неплохой набор свойств, особенно учитывая, что защищаемя структура данных может быть любой. Сценарий с редко обновляемой, зато часто читаемой структурой данных очень типичен, и обычно в нем используется что-то вроде CopyOnWrite — очень простой и надежный алгоритм, но требующий аллокации. Здесь же аллокаций после инициализации нет вообще (в самом алгоритме — если защищаемая структура данных нуждается в аллокациях при выполнении над ней каких-то операций, то эти аллокации, конечно, никуда не исчезнут). Тогда у меня не дошли руки описать их алгоритм, зато дошли сейчас :)

Off-heap: троллинг 70-го уровня

Последнее время я часто сталкиваюсь с идеей off-heap storage в java. Это когда DirectByteBuffer/Unsafe используют для организации данных вне java heap. Предположительно, это должно радикально снизить нагрузку на GC, да и, в некоторых случаях, сам доступ к данным ускорить. Сам я к этой идее относился довольно скептически: на мой взгляд у нее довольно узкая область применимости, и усилиями разработчиков JVM она становится с каждым годом еще уже. Мне кажется, многие, кто смотрит в сторону off-heap просто еще не разобрались толком, что можно делать на чистой java. Но пока это были все теоретические предположения — с Unsafe я проводил лишь один небольшой эксперимент, и, хоть предположения он, косвенно, и подтвердил, но одного эксперимента для выводов явно недостаточно. А сделать серьезный бенчмарк как-то было лениво.

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

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

Компоненты

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

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

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

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

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

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

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

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

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


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

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

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

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

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

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

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

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

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

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

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

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

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

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

Обработчики

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

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

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

Эпилог

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

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

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

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

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

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

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

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

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

LMAX Disruptor #1: Идеи

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

Развернутый связный список

Стыдно признаться, но я никогда еще в своей практике не использовал LinkedList. Не находилось оказии -- всегда оказывалось, что ArrayList либо в принципе подходит лучше, либо алгоритм можно так перестроить, что будет подходить лучше, либо в принципе он хуже, но на потребных размерах выигрывает. Последовательный доступ, большие накладные расходы памяти на хранение указателей в узлах, недружественность к кэшу и векторным инструкциям -- на мой взгляд, этого достаточно, чтобы 10 раз подумать, прежде чем его использовать. Тем не менее, никуда не деться -- вставка в случайное место для ArrayList дороже, чем для Linked уже начиная где-то с 500-1000 элементов (ну еще есть многопоточные приложения, где недружественность к кэшу становится преимуществом -- меньше конкуренция разных вычислительных ядер за строки кэша).

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

А сегодня обнаруживаю, что идею украли прямо из головы. И даже обозвать уже успели, и в википедии записали. Называется развернутый связный список (unrolled linked list). Неплохая статья на MSDN по этой структуре данных. Автор провел небольшое исследование производительности, и нашел, что развернутый список обходится (последовательно) примерно в 2.5 раза медленнее массива (обычный связный -- в 12 раз медленнее). На мой взгляд -- вполне разумный компромисс.

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

Lock-free BufferedQueue

Дали тестовое задание перед собеседованием -- написать буфферизированную очередь. Т.е. на вход метода append(T item) из нескольких потоков сыплются некие объекты, и, раз в какой-то интервал времени, задаваемый при создании, накопившиеся объекты сбрасываются списком в заранее заданный callback. Причем, понятно, очередь должна иметь как можно большую пропускную способность, т.е. засинхронизировать метод append -- это не выход.

Первая моя версия использовала ReadWriteLock -- неэксклюзивным ReadLock-ом защищался append, эксклюзивным WriteLock-ом -- flush. Но потом я взмедитнул, и подумал, что блокировки там вообще лишние. И получилась у меня lock-free версия вот такой:

public class LockFreeBufferedQueue<ItemType> {

private final ICallback<Collection<ItemType>> callback;
private final long delay;
private final TimeUnit timeUnit;

private final ScheduledExecutorService executor;

private final CallableVoid> flushBufferTask = new Callable<Void>() {
public Void call() {
assert ( pending.get() > 0 ) : "Call without being scheduled";
flushBuffer();
return null;
}
};

private final AtomicInteger pending = new AtomicInteger( 0 );

/**
* volatile чтобы обойтись без блокировки в методе isClosed()
*/
private volatile boolean closed = false;

private volatile ScheduledFuture<Void> scheduledTask = null;

/**
* lock-free Queue implementation
*/
private final ConcurrentLinkedQueue<ItemType> queue = new ConcurrentLinkedQueue<ItemType>();

public LockFreeBufferedQueue( final ICallback<Collection<ItemType>> callback,
final long delay,
final TimeUnit timeUnit,
final ScheduledExecutorService executor ) {        
this.callback = callback;
this.delay = delay;
this.timeUnit = timeUnit;
this.executor = executor;
}

/**
* Не специфицированно, что должен делать метод close(), поэтому мы делаем так:
* 
* Уже добавленные задачи будут выполнены немедленно, в текущем потоке, добавление
* новых приведет к IllegalStateException в методе append().
*/
public void close() {
if ( !closed ) {
closed = true;
if ( scheduledTask != null ) {
if ( scheduledTask.cancel( false ) ) {
flushBuffer();
}
scheduledTask = null;
}
//todo but if executor is external?
executor.shutdown();
}
}

/**
* @throws IllegalStateException если очередь уже закрыта вызовом close()
* @throws NullPointerException  если item == null
*/
public void append( final ItemType item ) {
if ( closed ) {
throw new IllegalStateException( "Queue was closed" );
}

queue.add( item );

if ( pending.getAndIncrement() == 0 ) {
schedule();
}
}

private void flushBuffer() {
final int size = pending.get();

assert ( !queue.isEmpty() ) : "Call with empty queue";
assert ( size > 0 ) : "Call with 0 pending";

final Collection<ItemType> buffer = new ArrayList<ItemType>( size );

for ( int i = 0; i < size; i++ ) {
buffer.add( queue.poll() );
}

if ( pending.getAndAdd( -size ) > size ) {
schedule();
}

//todo что делать с исключениями?
callback.call( buffer );
}

private void schedule() {
scheduledTask = executor.schedule( flushBufferTask, delay, timeUnit );
}

public boolean isClosed() {
return closed;
}
}
//=============================
public interface ICallback<T> {
public void call( final T arg );
}

Это первый мой опыт в самостоятельном написании lock-free алгоритмов на CAS-примитивах, поэтому написал я этот код часа за два, а медитировал над ним дня 3. Но косяков так и не нашел -- все должно работать как задумано. Увлекательное это занятие -- обдумывать корректность таких алгоритмов, мозг работает явно на повышенных оборотах