Left-Right

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

Simple concurrent caches

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

Есть у нас доменный объект CurrencyEntity:

public final class Entity {
 private final String isoCode;

 public Entity( final String isoCode ) {
  this.isoCode = isoCode;
 }

 @Override
 public boolean equals( final Object o ) {
  if( this == o ) {
   return true;
  }
  if( !(o instanceof Entity )) {
   return false;
  }

  final Entity entity = ( Entity ) o;

  if( !isoCode.equals( entity.isoCode ) ) {
   return false;
  }

  return true;
 }

 @Override
 public int hashCode() {
  return isoCode.hashCode();
 }
}

(важно, что объект immutable). На вход (из сети, например) программа получает Entity в виде String isoCode, но внутри, понятно, хочется оперировать type-safe представлением. Однако создавать каждый раз новый экземпляр Entity не хочется -- зачем нагружать лишний раз GC? Кроме того, известно, что реально различных isoCode не так уж много -- ну, скажем, штук 100-200. Вот как бы так их кэшировать, чтобы кэш и конкурентным был, и накладные расходы минимальны (иначе кэшировать смысла нет -- создать/удалить такой мелкий объект чего-то стоит, но не так уж дорого)?

Вариант решения:

public class FastConcurrentEntityCache{
 private static final Object FREE = null;

 private final Entity[] entries;

 public FastConcurrentEntityCache( final int size ) {
  checkArgument( size > 1, "size[%s] must be >1", size );

  entries = new Entity[size];
 }

 public Entity get( final String isoCode ) {
  final int hash = isoCode.hashCode() & 0x7fffffff;

  final int length = entries.length;
  final int index = hash % length;

  Entity cur = entries[index];

  if( cur == FREE ) {
   //cache miss
   return evaluateAndStore( isoCode, index );
  } else if( cur.isoCode.equals( isoCode ) ) {
   //cache hit
   return cur;
  } else {// already FULL, must probe

   // compute the double hash
   final int probe = 1 + ( hash % ( length - 2 ) );
   int probedIndex = index;
   for( int i = 0; i < entries.length; i++ ) {
    probedIndex -= probe;
    if( probedIndex < 0 ) {
     probedIndex += length;
    }
    cur = entries[probedIndex];
    if( cur == FREE ) {
     //cache miss
     return evaluateAndStore( isoCode, probedIndex );
    } else if( cur.isoCode.equals( isoCode ) ) {
     //cache hit
     return cur;
    }
   }
   //cache miss, and cache is full 
   throw new IllegalStateException( "Cache[size:"+entries.length+"] is full: ["+isoCode+"]" );
  }
 }

 private Entity evaluateAndStore( final String isoCode,
                                  final int index ) {
  final Entity entity = new Entity( isoCode );
  entries[index] = entity;
  return entity;
 }

 public void purge() {
  Arrays.fill( entries, null );
 }
}

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

Подробности

Вопрос: Почему это работает? Ответ: А почему бы и нет? ;)

Вообще-то стоило бы чуть подробнее остановиться на том, как именно оно работает. Обычно предполагается, что если объект уже есть в кэше, то при запросе именно он и вернется. Но в данном случае, из-за отсутствия синхронизации, состав кэша может разными потоками видиться по-разному, так что понятие "объект есть в кэше" становится не очень определенным. Так что работать это будет так: если объект есть в кэше, и виден текущему потоку -- вернется он. Если не виден (или вообще нет) -- будет создан новый экземпляр. То есть кэш не сохраняет идентичность (preserve identity) -- для одного и того же ключа isoCode в программе могут существовать несколько разных экземпляров Entity.

Второй важный момент это то, что Entity -- immutable. В многопоточном окружении запись entries[index] = entity -- это публикация ссылки на объект. Без дополнительной синхронизации (а где она? а нет ее!) это будет публикация через data race. И только особые гарантии JMM для immutable объектов дают нам возможность утверждать, что если другой поток увидел ссылку на Entity, то он увидит через нее полностью инициализированный объект.

Собственно, здесь две ключевых идеи. Во-первых, ослабление семантики -- отказ от принципа один тюбик в одни зубы "для каждого ключа значение создается строго однажды". Во-вторых, использование immutable объектов для безопасной публикации через data race. Эти две идеи в сумме и позволяют обойтись без синхронизации.

Что еще можно с этим сделать?

Приведенная реализация подразумевает работу в стационарном режиме: через какое-то время все isoCode будут программе знакомы, и соответствующие Entity закэшированы, и видимы всем потокам единообразно. Т.е. через какое-то время новые экземпляры Entity создаваться перестанут. При этом важно, чтобы мы изначально угадали с размером кэша.

Но это не единственный вариант использования. Бывает другая задача: множество объектов в принципе не ограничено, но известно, что чаще всего используются недавние объекты. Т.е. хочется LRU кэш. И такое тоже можно:

public class FastLRUEntityCache{
 private static final Object FREE = null;

 private final Entity[] entries;

 public FastLRUEntityCache( final int size ) {
  checkArgument( size > 1, "size[%s] must be >1", size );

  entries = new Entity[size];
 }

 public Entity get( final String isoCode ) {
  final int hash = isoCode.hashCode() & 0x7fffffff;

  final int length = entries.length;
  final int index = hash % length;

  Entity cur = entries[index];

  if( cur == FREE ) {
   //cache miss
   return evaluateAndStore( isoCode, index );
  } else if( cur.isoCode.equals( isoCode ) ) {
   //cache hit
   return cur;
  } else {// already FULL, replace
   return evaluateAndStore( isoCode, index );
  }
 }

 private Entity evaluateAndStore( final String isoCode,
                                  final int index ) {
  final Entity entity = new Entity( isoCode );
  entries[index] = entity;
  return entity;
 }

 public void purge() {
  Arrays.fill( entries, null );
 }
}

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

Вопрос: А если у меня объект не immutable? Ответ: Если он не immutable, но фактически не меняется -- можно просто создать обертку

public class Freezer<T>{
   public final T item;
   public Freezer(final T item){
    this.item = item;
   }
}

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

Вопрос: А можно сделать кэш resizeable? Ответ: Можно, конечно. Но придется делать поле entities volatile -- что несколько увеличит стоимость обращения к кэшу. Я не стал с этим возиться.

Для пуристов, поясняю: да, сам по себе volatile load на широко используемых архитектурах ничего дополнительного не стоит, в сравнении с обычным load. Но он запрещает некоторые оптимизации. Например:

for( over 9000 ){
  final String isoCode = ...;
  final Entity entity = cache.get(isoCode);
  ...
}

Если entries является final (да и просто non-volatile), то компилятор может вынести его чтение за цикл. Если же entries volatile -- оно будет считываться на каждой итерации. Т.е. мы можем сэкономить что-то вроде (L1 load - register load) * (over 9000) тактов :)

Философское заключение про data race

Как мы могли увидеть, существование в програме data race сам по себе не является приговором "программа некорректная". Но существование гонки, как правило, резко увеличивает размер графа состояний программы, количество возможных сценариев выполнения. Если это не было предусмотрено -- есть близкая к 100% вероятность упустить какие-то сценарии, нарушающие заложенные в код инварианты. Но если распоряжаться гонками аккуратно, то можно осмысленно создать ситуацию, где граф состояний даже с гонкой остается обозримым. И тогда можно увидеть возможности для вот таких вот простых и эффективных решений.

Это я так, типа, хвастаюсь :)

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

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

Without locks

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

Самый наглядный пример реализации -- ленивый кэш. Вот, например, простейший вариант -- кэшируем только результат последнего вызова:

public class SingleLRUWaitFreeCache<In, Out> implements ICache<In, Out> {
    private final Function<In, Out> function;

    private transient Entry<In, Out> recentUsed = null;

    public static <In, Out> SingleLRUWaitFreeCache<In, Out> create( final Function<In, Out> f ) {
        return new SingleLRUWaitFreeCache<In, Out>( f );
    }

    public SingleLRUWaitFreeCache( final Function<In, Out> function ) {
        checkArgument( function != null, "function can't be null" );
        this.function = function;
    }

    @Override
    public Out get( final In key ) {
        if( key == null ){
            throw new IllegalArgumentException("null keys are not supported");
        }
        final Entry<In, Out> lru = recentUsed;
        if ( lru != null && lru.key.equals( key ) ) {
            //cache hit
            return lru.value;
        } else {
            //cache miss
            final Out value = function.apply( key );
            recentUsed = new Entry<In, Out>( key, value );
            return value;
        }
    }

    @Override
    public void purge() {
        //not strictly correct since recentUsed is not volatile
        recentUsed = null;
    }
}

class Entry<Key, Value> {
    public final Key key;
    public final Value value;

    Entry( final Key key,
               final Value value ) {
        this.key = key;
        this.value = value;
    }
}

Дополнительные условия, при которых это работает:

1. Функция function -- идемпотентна, т.е. ее повторный вызов с теми же аргументами не производит никаких дополнительных изменений в системе, по сравнению с первым вызовом. Лучше всего, чтобы она вообще была "чистой" (pure)
2. function должно быть безопасно вызывать из нескольких потоков в произвольном порядке
3. In/Out должны быть immutable объектами

Что мы получаем в итоге?

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

Так же можно реализовывать кэши произвольного размера -- только размер должен быть известен заранее. Сейчас у меня в голове бродит реализация Scatter/Gatherer на этой же основе.

Потестировал на днях свою реализацию lock-free BufferedQueue. Неожиданно, но версия с блокировками всухую обходит CAS. Пока не могу понять, в чем дело -- грешу на ConcurrentLinkedQueue

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

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. Но косяков так и не нашел -- все должно работать как задумано. Увлекательное это занятие -- обдумывать корректность таких алгоритмов, мозг работает явно на повышенных оборотах