Distributed relaxed queue in remote memory access model designing and diagnostics of computer systems.pdf При разработке параллельных программ для вычислительных систем (ВС) одной из ключевых является задача синхронизации процессов (потоков), обращающихся к разделяемым (concurrent, thread-safe) структурам данных. Разделяемые структуры данных являются базовым элементом в параллельном программировании, поэтому эффективность синхронизации существенно влияет на время выполнения программ. Такие структуры должны обеспечивать доступ параллельных процессов (потоков) в произвольные моменты времени [1-3]. Синхронизация в ВС реализуется средствами блокировок (locks) и неблокируемых (nonblocking) структур данных. Блокировки обладают интуитивной семантикой и часто не менее эффективны по сравнению с неблокируемыми методами. В то же время программирование без блокировок позволяет избежать тупиковых ситуаций (deadlocks), инверсий приоритетов (priority inversion) и обеспечивает гарантии выполнения. Независимо от реализации большая часть структур данных характеризуется наличием узких мест (bottlenecks) для операций, таких как вставка и удаление элементов (очереди, стеки), удаление максимального элемента (очереди с приоритетом). Большая часть работ в области разделяемых структур данных ориентированы на ВС с общей памятью, производительность которых может быть недостаточной для решения современных задач. Например, размеры графов социальных сетей достигают нескольких петабайт, а число вершин в графах из теста Graph500 - нескольких триллионов. Проекты в области физики высоких энергий, такие как CMS и Atlas, производят десятки петабайт ежегодно. Планируется, что Большой обзорный телескоп будет каждую ночь генерировать около 20 терабайт. Поскольку ВС с общей памятью имеют технологический предел числа процессорных ядер, для решения таких задач требуется использовать ВС с распределенной памятью (кластерные ВС, ВС с массовым параллелизмом). В процессе программирования таких систем обрабатываемые данные представляются в виде распределенных структур данных, для которых необходимо обеспечить масштабируемую синхронизацию. Одной из перспективных моделей параллельного программирования для ВС с распределенной памятью является модель удаленного доступа к памяти (Remote Memory Access, RMA), реализованная в стандарте MPI [4, 5]. В рамках RMA процессы непосредственно обращаются к памяти других процессов вместо отправки и получения сообщений. В отличие от модели разделенного глобального адресного пространства (Partitioned Global Address Space, PGAS), представленной языками UPC, CAF, Chapel, X10, модель RMA тесно интегрирована с библиотеками MPI и может быть использована наравне с моделью передачи сообщений. Программы в модели RMA характеризуются меньшим временем выполнения по сравнению с моделью передачи сообщений и PGAS. Большая часть современных коммуникационных сетей (Infiniband, PERCS, Gemini, Aries, RoCE over Ethernet) обеспечивает поддержку RMA с помощью технологии RDMA [4], реализующей обращение к удаленным сегментам памяти без участия центрального процессора. Опишем программную модель RMA в MPI. Основными являются неблокируемые функции MPI_Put (запись в память удаленного процесса) и MPI_Get (чтение из удаленной памяти), атомарные MPI_Accumulate, MPI_Get_accumulate, MPI_Compare_and_swap. RMA-вызовы должны находиться внутри областей (эпохи, epochs), в рамках которых выполняется синхронизация. В работе применяется пассивный метод синхронизации (passive target synchronization), реализованный в стандарте MPI [5]. При пассивной синхронизации процесс открывает эпоху реализации удаленного доступа (access epoch) посредством вызова функций MPI_Win_lock/MPI_Win_lockall, после чего он может выполнять RMA-операции для доступа к зарегистрированным сегментам памяти (окна, windows) других процессов. Таким образом, RMA-операции выполняются в одностороннем порядке, без явного вызова функций синхронизации другими процессами. Основная часть работ в области разделяемых структур данных направлена на создание средств синхронизации для ВС с общей памятью. К ним относятся алгоритмы блокировки потоков [1, 6] (TTS, Backoff, CLH, MCS, Oyama, Flat Combining, RCL и др.). Хотя некоторые методы (Hierarchical Backoff (CLH, MCS), Cohorting и др.) учитывают отдельные иерархические уровни, они неприменимы в ВС с распределенной памятью. Неблокируемые структуры [1-3, 7] также разработаны для многоядерных ВС и неприменимы в распределенных ВС. Перспективным методом повышения масштабируемости разделяемых структур данных является ослабление их семантики (relaxation) [8-14]. Например, в ослабленной очереди с приоритетом извлекается не максимальный элемент, а элемент, близкий к максимальному. В ослабленной очереди (стеке) удаляется не первый (последний) добавленный элемент, а элемент в его окрестности. Ослабленные структуры обеспечивают высокую пропускную способность и приемлемый уровень упорядоченности операций в реальных программах. В работах [8, 9] для построения потокобезопасной очереди с приоритетом предлагается использовать набор последовательных очередей. Аналогичная реализация стека, основанного на временных метках, предлагается в [10]. Также построены аналитические модели ослабления [13, 14], включая квазилинеаризуемость (quasi linearizability), количественное ослабление (quantitative relaxation). Насколько известно, для распределенных ВС не разработаны эффективные масштабируемые разделяемые структуры данных. Методы, предложенные для распределенных ВС, включают простые спинлоки, блокировки чтения-записи и MCS-блокировки [15]. Работы, посвященные распределенным структурам данных [16-18], неприменимы в модели RMA. В языках PGAS реализованы отдельные примитивы синхронизации и распределенные структуры, но они характеризуются наличием узких мест и высокими накладными расходам. Исходя из вышесказанного, задача разработки эффективных разделяемых структур данных для распределенных ВС является востребованной и нерешенной в настоящее время. В данной статье предлагается метод построения масштабируемых распределенных структур на основе ослабления их семантики, рассмотренный на примере очереди. 1. Распределенная ослабленная очередь 1.1. Ослабление семантики выполнения операций для распределенных очередей Очередь - коллекция объектов, реализующая дисциплину FIFO («первым вошел - первым вышел»). Основные операции: добавление (insert, enqueue) элемента в последнюю позицию (хвост, tail) и извлечение (remove, dequeue) элемента из первой позиции (голова, head). В классических реализациях распределенных очередей необходимо обеспечивать актуальность данных о расположении (ранг процесса) головного и хвостового элементов. Процесс перед выполнением операций при необходимости обновляет данные о расположении головного (хвостового) элемента. Это приводит к дополнительным накладным расходам и увеличивает время выполнения операций. Другим значимым недостатком является наличие узких мест при одновременном обращении нескольких процессов к процессу, в памяти которого находится головной (хвостовой) элемент. С целью увеличения масштабируемости распределенной очереди предлагается ослабить ее семантику и допустить извлечение элемента из окрестности первого добавленного элемента. Для этого распределенная структура представляется в виде множества последовательных структур, распределенных между процессами. Каждый процесс асинхронно обращается к удаленным сегментам посредством RMA-вызовов (рис. 1). Данный подход не предполагает выполнения операций для актуализации данных о расположении головы и хвоста очереди и позволяет избежать возникновения узких мест. Кроме того, за счет низкой латентности односторонних коммуникаций и аппаратной поддержки RDMA метод гарантирует снижение времени выполнения операций. Гроцесс 1 Процеос 2 Процесс n Рис. 1. Операция извлечения элемента в ослабленной распределенной очереди Fig. 1. Execution of item remove operation for relaxed distributed queue Опишем операции добавления и удаления элементов. Обозначимр - число процессов. Считаем, что часы процессов синхронизированы и каждому элементу e очереди соответствует временная метка t(e) - момент добавления его в очередь. При выполнении операции insert процесс случайным образом выбирает очередь s е {1, 2, ...,р} и помещает в нее элемент. Отметим, что вместо случайного выбора можно использовать другие схемы для локализации обращений к памяти и других оптимизаций. При выполнении операции удаления remove процесс выбирает k очередей-кандидатов R с {1, 2, ...,р} и посредством RMA-операций получает значения элементов (ец e2, ..., ek}, находящихся в голове соответствующих очередей. Далее среди элементов-кандидатов определяется «лучший» элемент с минимальной временной меткой: e* = argmin е {1, 2, ...,k} t(ei) (рис. 1). Рассмотрим детально реализацию распределенной ослабленной очереди. При инициализации структуры данных на каждом процессе организуется циклический буфер фиксированного достаточно большого размера (100 000 в данной реализации) и синхронизируются часы процессов [19]. 1.2. Операция добавления элементов Входными данным операции insert добавления элемента являются значение элемента val, число процессов р, окно для выполнения RMA-операций win и массив блокировок locks для защиты данных в очередях на каждом процессе. Блокировки могут быть реализованы с помощью любого спинлока, в данной работе используется простой алгоритм TASLock [1]. Под MPI_Put_Atomic и MPI_Get_atomic здесь и далее понимаются атомарные операции MPI_Put и MPI_Get, реализованные на основе MPI_Accumulate и MPI_Get_accumulate. Т а б л и ц а 1 Алгоритмы выполнения операций для распределенной ослабленной очереди: а - операция insert добавления элемента в очередь, б - операция remove извлечения элемента из очереди а b Входные val - добавляемый элемент данные: locks - массив блокировок для защиты очередей p - число процессов коммуникатора win - окно для выполнения RMA-операций Входные ncand - число очередей-кандидатов данные: locks - массив блокировок для защиты очередей p - число процессов коммуникатора win - окно для выполнения RMA-операций i nqueues=p 1 MPI_WlN_LOCK_ALL(win) 2 do 2 ncurr = 0 3 rank = GetRand(p) 3 navail = p 4 elem.val = val 4 nattempts = 0 5 elem.ts = GetTimeStamp() 5 while ncurr < ncand do 6 MPI Win lock(rank, win) 6 rank = GetRand(p) 7 LOCk(rank, locks[rank], win) 7 rc = TRYLOCK(rank, locks[rank], win) 8 MPI Get ATOMiC(rank, state, win) 8 if LOCKlSACQUiRED(rc) then 9 MPI_W in_flush( win) 9 MPI Get ATOMiC(rank, states[ncurr], win) 10 if iSFULL(state) then 10 MPI_WlN_FLUSH(win) 11 nqueues = nqueues - 1 11 if iSEMPTY(state) then 12 if nqueues = 0 then 12 UnLock(rank, locks[rank], win) 13 UNLOCK(rank, locks[rank], win) 13 navail = navail - 1 14 MPI Win unlock(rank, win) 14 if navail < ncand then 15 return ErrQueueFull 15 if ncand = 0 then 16 end if 16 MPI_WlN_UNLOCK_ALL(win) 17 else 17 return ErrQueueEmpty 18 MPI PUT(rank, elem, win) 18 end if 19 state.tail = (state.tail + 1) mod size 19 ncand = navail 20 MPI Put ATOMiC(rank, state.tail, win) 20 end if 21 end if 21 else 22 UNLOCK(rank, locks, win) 22 MPI GET(rank, elems[ncurr], win) 23 MPI Win unlock(rank, win) 23 MPI_WlN_FLUSH(win) 24 while iSFuLL(state) 24 ADDCAND(rank, cands) 25 ncurr = ncurr + 1 26 nattempts = 0 27 end if 28 else if LOCKlSBuSY(rc) then 29 if ncurr > 0 then 30 nattempts = nattempts + 1 31 if nattempts = max nattempts then 32 for i = 0 to ncurr do 33 UNLOCK(cands[i], locks[cands[i]], win) 34 end for 35 ncurr = 0 36 end if 37 end if 38 end if 39 end while 40 bestrank = GetBestRank(cands, elems) 41 states[bestrank].tail = (states[bestrank].tail + 1) mod 42 size 43 MPI Put ATOMiC(bestrank, states[bestrank].tail, win) 44 for i = 0 to ncand do 45 UNLOCK(cands[i], locks[cands[i]], win) 46 end for 47 MPI_WlN_UNLOCK_ALL(win) return elems[bestrank].val Основные шаги алгоритма (табл. 1, а): 1. Проинициализировать число доступных очередей nqueues (строка 1). 2. Случайным образом выбрать процесс rank (строка 3). Установить поля элемента (строки 4, 5). Начать эпоху синхронизации для выбранного процесса (строка 6). Заблокировать очередь процесса rank (строка 7) и получить ее состояние (строка 8). 3. Если очередь заполнена (строка 10), уменьшить nqueues. Если нет доступных очередей (строка 12), разблокировать очередь, завершить эпоху синхронизации, вернуть код ошибки (строки 13-15). 4. Если очередь не полна, добавить в нее элемент (строка 18), увеличить указатель state.tail на хвост очереди (строка 19) и установить новое значение состояния очереди (строка 20). 5. Разблокировать очередь (строка 22) и завершить эпоху пассивной синхронизации (строка 23). 6. Выполнять шаги 2-5, пока как минимум одна очередь не будет найдена (строка 24). 1.3. Операция удаления элементов Входными данными для функции remove удаления элемента являются число кандидатов ncand для выбора элемента, количество процессов p, окно для выполнения RMA-операций win и массив блокировок locks для защиты данных очередей. Операция включает следующие шаги (табл. 1, b): 1. Начать эпоху пассивной синхронизации для всех процессов (строка 1), проинициализировать текущее число найденных очередей ncurr (строка 2), число доступных очередей navail (строка 3) и число попыток блокировки очереди nattempts (строка 4). 2. Если текущее число найденных кандидатов ncurr равно ncand, перейти на шаг 7. Если нет, случайно выбрать очередь rank (строка 6). Попытаться заблокировать очередь (строка 7). 3. Если очередь заблокирована, получить ее состояние state (строка 9). Если нет, переход на шаг 6. 4. Если очередь пуста (строка 11), разблокировать ее (строка 12), уменьшить navail (строка 13). Если navail < ncand, сбросить ncand до значения navail. Если не осталось доступных кандидатов (строка 15), завершить эпоху синхронизации (строка 16) и вернуть код ошибки (строка 17). 5. Если очередь не пуста, получить элемент в голове (строка 22), добавить в список кандидатов (строка 24), увеличить ncurr (строка 25), сбросить nattempts (строка 26). Перейти на шаг 2. 6. Если очередь не захвачена, увеличить nattempts (если это не первая очередь-кандидат) (строка 30). Если nattempts достигло максимального (строка 31), разблокировать захваченные очереди (строки 32-34), сбросить ncurr (строка 35), перейти на шаг 2. Данный шаг необходим для избежания взаимной блокировки, когда два процесса пытаются заблокировать уже захваченные очереди. 7. Выбрать кандидата с минимальным значением временной метки (строка 40). Для данной очереди инкрементировать указатель на голову и обновить состояние очереди (строка 41, 42). Разблокировать все очереди-кандидаты (строки 43, 45) и завершить эпоху синхронизации (строка 46). 2. Проведение экспериментов Экспериментальное исследование ослабленной очереди проводилось на вычислительном кластере Jet Центра параллельных вычислительных технологий Сибирского государственного университета телекоммуникаций и информатики. Кластер укомплектован 18 вычислительными узлами, оборудованными двумя 4-ядерными процессорами Intel Xeon E5420 (суммарное число ядер 144). В качестве MPI-библиотеки применялась MPICH 3.2.1. Разработан синтетический тест, выполняющий n = 100 000 операций вставки / удаления (тип операции выбирается случайно). Число процессов p варьировалось от 16 до 144. Созданная распределенная ослабленная очередь Relaxed Queue сравнивалась со связным списком, реализованным в библиотеке MPICH (MPICH linked list) в модели RMA. Использовалось два типа списка: на основе эксклюзивной и разделяемой пассивной синхронизации (MPI_LOCK_EXCLUSIVE и MPI_LOCK_SHARED). Тип синхронизации определяет, допускается ли одновременное обращение нескольких процессов к памяти удаленного процесса. Также исследовалось влияние числа очередей-кандидатов ncand на эффективность очереди. Для этого ncand варьировалось от 1 до 4. Кроме того, анализировалась зависимость эффективности очереди от типа пассивной синхронизации в функции вставки. Измерялась пропускная способность b = t / n, где t - время проведения эксперимента. Пропускная способность разработанной очереди значительно превосходит пропускную способность линейного списка строгой семантики (рис. 2, а). Оптимизация достигается за счет сокращения накладных расходов, возникающих при выполнении доступа к элементам. Недостатки классических распределенных списков - необходимость актуализации данных о расположении головного (хвостового) элементов и возможность образования узких мест при одновременном обращении нескольких процессов к ним. Разработанная ослабленная очередь не требует поддержания согласованного состояния головы (хвоста) очереди, поскольку каждый раз процесс-кандидат и соответствующая очередь выбираются случайно. Такой подход также позволяет распределить нагрузку между процессами и избежать возникновения узких мест. Линейный список строгой семантики на основе разделяемой синхронизации более эффективен, по сравнению с эксклюзивным режимом (рис. 2, b), поскольку во время вставки / удаления несколько процессов одновременно обращаются к одному процессу, в памяти которого находится головной (хвостовой) элемент. b, опер/с а b Рис. 2. Сравнение эффективности структур данных: а - пропускная способность; b - пропускная способность линейного списка строгой семантики для разных режимов синхронизации Fig. 2. Comparison of efficiency of data structures: a - throughput; b - throughput of list with strong semantics for different synchronization modes b, опер/с A-*b, опер/с а b Рис. 3. Анализ эффективности Relaxed Queue: а - пропускная способность в зависимости от числа очередей-кандидатов; б - пропускная способность ослабленной очереди для разных режимов синхронизации Fig. 3. Analysis of Relaxed Queue efficiency: a - throughput depending on number of candidates; b - throughput of Relaxed Queue for different synchronization modes Как и ожидалось, пропускная способность уменьшается с увеличением числа кандидатов ncand (рис. 3, а). Это объясняется дополнительными накладными расходами на выполнение блокировки очередей, получение состояний и значений элементов очередей-кандидатов. На наш взгляд, ncand = 2 является достаточным для большинства случаев и обеспечивает приемлемый для практики уровень упорядоченности операций. Данные выводы согласуются с результатами аналогичной структуры для ВС с общей памятью [8]. Тем не менее для повышения близости порядка вставки / удаления элементов к порядку FIFO можно увеличить ncand до 3 и 4. В данной работе не выполняется оценка близости к FIFO, но это планируется сделать в будущем. В отличие от распределенных списков со строгой семантикой, для ослабленной очереди эксклюзивный режим пассивной синхронизации обеспечивает большую пропускную способность по сравнению с разделяемым режимом (рис. 3, b). Это объясняется тем, что организация разделяемого режима является дорогостоящей операцией. Вместе с тем в ослабленной очереди одновременное обращение к одному процессу (последовательной очереди) является редким событием. Поэтому мы полагаем, что разделяемый режим является избыточным. Заключение В данной статье разработаны эвристические алгоритмы реализации ослабленных распределенных очередей в модели RMA. Созданная очередь основана на множестве последовательных очередей, распределенных между процессами. Очередь характеризуется значительно большей пропускной способностью по сравнению с линейными списками строгой семантики в модели RMA. Оптимизация достигается за счет устранения узких мест при выполнении операций. При реализации очереди рекомендуется использовать 2 или 3 очереди-кандидата и эксклюзивный тип пассивной синхронизации (MPI_LOCK_EXCLU SIVE).

Скачать электронную версию публикации