Идея такая: что очередь -- это простой линейный массив (универсальная штука этот массив). Каждый поток-производитель выбирает себе произвольный (случайный) стартовый индекс в этом массиве, и кладет в очередь элементы так:
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, что у действительно многоядерных систем встречается гораздо чаще -- тогда, мне кажется, все будет вообще плохо, потому что массив будет скорее всего выделен в локальной памяти инициализирующего потока, а использоваться будет всеми потоками.
А как CAS-то на массиве делать в java? Заполнять его AtomicReference или есть более красивая возможность?
ОтветитьУдалитьНу для простых смертных есть AtomicReferenceArray. Но Клиф -- он небожитель, он использует напрямую sun.misc.Unsafe. Все AtomicXXX реализованы (в Sun jdk, и всех его производных -- т.е. практически везде) через вызовы Unsafe, а вызовы Unsafe являются, насколько я знаю, интринсиками JIT-компилятора. Одним из интересных возможностей Unsafe является то, что через него можно делать волатильное чтение/запись и CAS обычных, неволатильных полей. Этим-то Клиф регулярно и пользуется.
ОтветитьУдалитьВ общем, чтобы понять что именно он делает -- можно просто посмотреть исходники AtomicReferenceArray
Спасибо, не знал, что есть прямо обертка для массива.
ОтветитьУдалитьКстати, "Одним из интересных возможностей Unsafe является то, что через него можно делать волатильное чтение/запись и CAS обычных, неволатильных полей." А об этом где можно почитать по-подробнее. Просто в этом случае же получается, что те же атомики из java.util.concurrent.atomic можно намного эффективнее написать, в библиотечных то поле, содержащее, значение является volatile. Таким образом метод get будет сильно медленнее чем мог бы быть. Конечно, если убрать volatile, то придется и метод set с конструктором на Unsafe переписывать, но конструктор один раз вызывается, так что не проблема, а метод set без свопа у атомиках, думаю, не сильно популярен.
Или ты имеешь ввиду метод sun.misc.Unsafe#putXXXVolatile и sun.misc.Unsafe#getXXXVolatile? Думаешь они быстрее обращений к обычным volatile переменным? Если же бы это было так, то JIT, ведь, бы всю эту магию всегда проделывал...?
ОтветитьУдалить>А об этом где можно почитать по-подробнее
ОтветитьУдалитьПодробнее -- собственно, в исходниках AtomicXXX -- смотреть использование. Либо скачивать полные исходники JDK и смотреть сам Unsafe с комментариями к методам. Но в общем-то это не особо полезно
>Просто в этом случае же получается, что те же атомики из java.util.concurrent.atomic можно намного эффективнее написать
Нет, нельзя :) Если get() не будет содержать волатильного чтения -- большая часть его использований будет некорректным, ибо по JMM рантайм будет вообще не обязан хоть когда-нибудь считывать из оперативки его актуальное значение, он будет иметь право закэшировать его в регистрах хоть навсегда. В частности, какой-нибудь спин-лок с Test-and-CAS может легко вообще зависнуть навсегда.
На самом деле переписывать работу с атомиками через Unsafe очень редко когда имеет смысл. В большинстве случаев код написанный через AtomicXXX JIT заоптимизирует до такого же состояния сам -- там же большинство методов по-факту мономорфные, а значит будут заинлайнены с вероятностью близкой к наверняка.
Клиф использует Unsafe не столько из-за производительности (хотя это я так думаю), сколько из-за того, что у Unsafe контракт слабее, чем у AtomicXXX, и это позволяет ему играть тонко.
Сорри, поздно уже, нверное, поэтому туплю. Так что можешь удалить мои комменты. Это же дает возможность исползовать volatile чтение и запись только в определнных случах, а во всех остальных, допустим, использовать прямой доступ. Хотя так сходу пример даже и не придумаю :)
ОтветитьУдалитьДа и спасибо, наконец-то я нашел способ писать/читать волатильно конкретный элемент final массива без специальных оберток.
>Хотя так сходу пример даже и не придумаю
ОтветитьУдалитьПример можно придумать, но он будет своеобразным. Тот же самый безблокировочный кэш, о котором я писал несколько месяцев назад. Там метод purge() неизвестно когда толком сработает -- т.е. значения могут застревать в кэше надолго. Это не влияет на семантику, но ухудшает производительность. так вот если там использовать вместо обычной записи Unsafe -- время застревания уменьшится :)
>наконец-то я нашел способ писать/читать волатильно конкретный элемент final массива без специальных оберток.
Сие грешно делать помимо Клифа и Дага Ли!
А чем собственно AtomicReferenceArray не устраивал?
Совершенно согласен! Чисто академический интерес. Да и не знал я про AtomicReferenceArray ...
ОтветитьУдалить