Многопроцессорная обработка и распределённые вычисления

Реализация параллельных вычислений с распределённой памятью предоставляется модулем Distributed в качестве части стандартной библиотеки Julia.

Большинство современных компьютеров имеют более одного процессора, и несколько компьютеров могут быть объединены в кластер. Использование мощности нескольких процессоров позволяет завершить многие вычисления быстрее. На производительность влияют два основных фактора: скорость самих процессоров и скорость доступа к памяти. В кластере очевидно, что любой процессор имеет наибыстрейший доступ к оперативной памяти в том же компьютере (узле). Возможно, несколько неожиданно, что аналогичные проблемы актуальны для типичного многоядерного ноутбука из-за различий в скорости оперативной памяти и кеша. Следовательно, хорошая многопроцессорная среда должна позволять управлять «владением» блоком памяти конкретным процессором. Julia предоставляет многопроцессорную среду на основе обмена сообщениями, позволяющую программам выполняться на нескольких процессах в отдельных областях памяти одновременно.

Реализация обмена сообщениями в Julia отличается от других сред, таких как MPI^[1]. Обмен сообщениями в Julia, как правило, является «односторонним», что означает, что программист должен явно управлять только одним процессом в операции с двумя процессами. Кроме того, эти операции обычно не похожи на «отправку сообщения» и «приём сообщения», а скорее напоминают операции более высокого уровня, такие как вызовы пользовательских функций.

Распределённое программирование в Julia основано на двух примитивах: удалённых ссылках и удалённых вызовах. Удалённая ссылка — это объект, который может использоваться любым процессом для ссылки на объект, хранящийся на конкретном процессе. Удалённый вызов — это запрос одного процесса для вызова определённой функции с определёнными аргументами на другом (возможно, на том же) процессе.

Удалённые ссылки бывают двух типов: Future и RemoteChannel.

Удалённый вызов возвращает Future результата. Удалённые вызовы возвращаются немедленно; процесс, который сделал вызов, переходит к следующей операции, в то время как удалённый вызов происходит где-то ещё. Вы можете дождаться завершения удалённого вызова, вызвав wait на возвращённом Future, и вы можете получить полное значение результата, используя fetch.

С другой стороны, RemoteChannel можно перезаписывать. Например, несколько процессов могут координировать свою обработку, ссылаясь на один и тот же удалённый Channel.

Каждый процесс имеет связанный идентификатор. Процесс, предоставляющий интерактивный приглашающий интерфейс Julia, всегда имеет id равный 1. Процессы, используемые по умолчанию для параллельных операций, называются «рабочими». Когда существует только один процесс, процесс 1 считается рабочим. В противном случае рабочими считаются все процессы, кроме процесса 1. В результате требуется добавление 2 или более процессов, чтобы получить преимущества от методов параллельной обработки, таких как pmap. Добавление одного процесса полезно, если вы просто хотите сделать что-то другое в основном процессе, в то время как длинное вычисление выполняется на рабочем процессе.

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

$ julia -p 2

julia> r = remotecall(rand, 2, 2, 2)
Future(2, 1, 4, nothing)

julia> s = @spawnat 2 1 .+ fetch(r)
Future(2, 1, 5, nothing)

julia> fetch(s)
2×2 Array{Float64,2}:
 1.18526  1.50912
 1.16296  1.60607

Первый аргумент функции remotecall — функция для вызова. Большинство параллельных вычислений в Julia не ссылаются на конкретные процессы или количество доступных процессов, но remotecall считается низкоуровневым интерфейсом, обеспечивающим более точный контроль. Второй аргумент remotecall — id процесса, который выполнит работу, а оставшиеся аргументы будут переданы вызываемой функции.

Как видите, в первой строке мы попросили процесс 2 создать случайную матрицу 2x2, а во второй строке — добавить к ней 1. Результат обоих вычислений доступен в двух будущих значениях, r и s. Макрос @spawnat вычисляет выражение во втором аргументе на процессе, указанном в первом аргументе.

Иногда вам может потребоваться немедленно получить удалённо вычисленное значение. Это обычно происходит, когда вы читаете удалённый объект для получения данных, необходимых для следующей локальной операции. Функция remotecall_fetch существует для этой цели. Она эквивалентна fetch(remotecall(...)), но более эффективна.

julia> remotecall_fetch(r-> fetch(r)[1, 1], 2, r)
0.18526337335308085

Это извлекает массив из рабочего процесса 2 и возвращает первое значение. Обратите внимание, что fetch в этом случае не перемещает данные, так как он выполняется на рабочем процессе, который владеет массивом. Также можно написать:

julia> remotecall_fetch(getindex, 2, r, 1, 1)
0.10824216411304866

Помните, что getindex(r,1,1) эквивалентно r[1,1], поэтому этот вызов извлекает первый элемент из будущего значения r.

Для упрощения символ :any можно передать в @spawnat, который выберет для вас место выполнения операции:

julia> r = @spawnat :any rand(2,2)
Future(2, 1, 4, nothing)

julia> s = @spawnat :any 1 .+ fetch(r)
Future(3, 1, 5, nothing)

julia> fetch(s)
2×2 Array{Float64,2}:
 1.38854  1.9098
 1.20939  1.57158

Обратите внимание, что мы использовали 1 .+ fetch(r) вместо 1 .+ r. Это потому, что мы не знаем, где будет выполняться код, поэтому в общем случае может потребоваться fetch для перемещения r на процесс, выполняющий сложение. В этом случае @spawnat достаточно умён, чтобы выполнить вычисление на процессе, который владеет r, поэтому fetch будет недействительной операцией (работа не выполняется).

(Стоит отметить, что @spawnat не встроен, а определён в Julia как макрос. Можно определить собственные подобные конструкции.)

Важный момент: после извлечения Future кеширует своё значение локально. Дальнейшие вызовы fetch не предполагают сетевой обмен. После извлечения всех ссылающихся Future, удалённое сохранённое значение удаляется.

@async похож на @spawnat, но выполняет задачи только на локальном процессе. Мы используем его для создания задачи «подачи» для каждого процесса. Каждая задача выбирает следующий индекс, который нужно вычислить, затем ждёт завершения своего процесса, затем повторяет, пока не закончатся индексы. Обратите внимание, что задачи подачи не начинают выполняться, пока основная задача не достигнет конца блока @sync, в котором она уступает управление и ждёт завершения всех локальных задач, прежде чем возвращаться из функции. Что касается версии v0.7 и выше, задачи подачи могут обмениваться состоянием через nextidx, так как они все выполняются на одном процессе. Даже если Tasks планируются кооперативно, в некоторых контекстах всё ещё может потребоваться блокировка, как в случае с асинхронным вводом-выводом. Это означает, что переключения контекста происходят только в определённых точках: в данном случае, когда вызывается remotecall_fetch. Это текущее состояние реализации, и оно может измениться в будущих версиях Julia, так как предполагается, что это позволит запускать до N Tasks на M Process, т. е. M:N потоков. Затем потребуется модель приобретения/освобождения блокировки для nextidx , так как небезопасно допускать одновременное чтение-запись ресурса несколькими процессами.

Доступность кода и загрузка пакетов

Ваш код должен быть доступен на любом процессе, который его запускает. Например, введите следующее в приглашении Julia:

julia> function rand2(dims...)
           return 2*rand(dims...)
       end

julia> rand2(2,2)
2×2 Array{Float64,2}:
 0.153756  0.368514
 1.15119   0.918912

julia> fetch(@spawnat :any rand2(2,2))
ERROR: RemoteException(2, CapturedException(UndefVarError(Symbol("#rand2"))))
Stacktrace:
[...]

Процесс 1 знал о функции rand2, но процесс 2 — нет.

Чаще всего вы будете загружать код из файлов или пакетов, и у вас есть значительная гибкость в управлении загрузкой кода на различных процессах. Рассмотрим файл DummyModule.jl, содержащий следующий код:

module DummyModule

export MyType, f

mutable struct MyType
    a::Int
end

f(x) = x^2+1

println("loaded")

end

Для того, чтобы ссылаться на MyType через все процессы, DummyModule.jl нужно загрузить на каждый процесс. Вызов include("DummyModule.jl") загружает его только на один процесс. Для загрузки на каждый процесс используйте макрос @everywhere (начало Julia с julia -p 2):

julia> @everywhere include("DummyModule.jl")
loaded
      From worker 3:    loaded
      From worker 2:    loaded

Как обычно, это не помещает DummyModule в область видимости ни на одном из процессов, что требует using или import. Более того, когда DummyModule попадает в область видимости на одном процессе, его нет ни на каком другом:

julia> using .DummyModule

julia> MyType(7)
MyType(7)

julia> fetch(@spawnat 2 MyType(7))
ERROR: On worker 2:
UndefVarError: `MyType` not defined in `Main`
⋮

julia> fetch(@spawnat 2 DummyModule.MyType(7))
MyType(7)

Однако по-прежнему возможно, например, отправить MyType на процесс, на котором загружен DummyModule , даже если он не находится в области видимости:

julia> put!(RemoteChannel(2), MyType(7))
RemoteChannel{Channel{Any}}(2, 1, 13)

Файл также можно предварительно загрузить на несколько процессов во время запуска с флагом -L, а скрипт драйвера можно использовать для управления вычислением:

julia -p <n> -L file1.jl -L file2.jl driver.jl

Процесс Julia, выполняющий скрипт драйвера в приведённом примере, имеет id равный 1, как и процесс, предоставляющий интерактивное приглашение.

Наконец, если DummyModule.jl не является автономным файлом, а пакетом, тогда using DummyModule загрузит DummyModule.jl на все процессы, но поместит его в область видимости только на том процессе, где был вызван using.

Запуск и управление рабочими процессами

Базовая установка Julia имеет встроенную поддержку двух типов кластеров:

• Локальный кластер, указанный с помощью опции -p, как показано выше.
• Кластер, охватывающий машины, использующий опцию --machine-file. Для запуска процессов Julia worker (из той же директории, что и текущий хост) на указанных машинах используется бесклеточная ssh авторизация. Описание каждой машины имеет вид [count*][user@]host[:port] [bind_addr[:port]]. user по умолчанию равен текущему пользователю, port — стандартному порту SSH. count — количество процессов worker, которое необходимо запустить на узле; по умолчанию значение равно 1. Необязательный параметр bind-to bind_addr[:port] определяет IP-адрес и порт, которые другие worker должны использовать для подключения к данному worker.

Функции addprocs, rmprocs, workers и другие доступны в качестве программируемых средств для добавления, удаления и запроса процессов в кластере.

julia> using Distributed

julia> addprocs(2)
2-element Array{Int64,1}:
 2
 3

Модуль Distributed должен быть явно загружен на главном процессе перед вызовом addprocs. Он автоматически становится доступным для процессов worker.

Другие типы кластеров могут быть поддержаны написанием собственных пользовательских ClusterManager, как описано ниже в разделе ClusterManagers.

Перемещение данных

Отправка сообщений и перемещение данных составляют большую часть накладных расходов в распределённой программе. Снижение количества сообщений и объёма отправляемых данных критически важно для достижения производительности и масштабируемости. Для этого важно понимать перемещение данных, выполняемое различными распределёнными конструкциями Julia.

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

Метод 1:

julia> A = rand(1000,1000);

julia> Bref = @spawnat :any A^2;

[...]

julia> fetch(Bref);

Метод 2:

julia> Bref = @spawnat :any rand(1000,1000)^2;

[...]

julia> fetch(Bref);

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

В этом примере два метода легко различить и выбрать. Однако в реальной программе проектирование перемещения данных может потребовать большего размышления и, вероятно, некоторых измерений. Например, если первому процессу нужна матрица A, то первый метод может быть лучше. Или, если вычисление A дорогостоящее, и только текущий процесс обладает им, то перемещение его в другой процесс может быть неизбежным. Или, если текущий процесс имеет очень мало работы между @spawnat и fetch(Bref), то может быть лучше вообще исключить параллелизм. Или представьте, что rand(1000,1000) заменено более дорогостоящей операцией. Тогда может иметь смысл добавить ещё одну инструкцию @spawnat только для этого шага.

Глобальные переменные

Выражения, выполняемые удалённо с помощью @spawnat, или замыкания, заданные для удалённого выполнения с помощью remotecall, могут ссылаться на глобальные переменные. Глобальные привязки в модуле Main обрабатываются немного иначе по сравнению с глобальными привязками в других модулях. Рассмотрим следующий фрагмент кода:

A = rand(10,10)
remotecall_fetch(()->sum(A), 2)

В этом случае sum ДОЛЖЕН быть определён в удалённом процессе. Обратите внимание, что A — это глобальная переменная, определённая в локальном пространстве имен. Worker 2 не имеет переменной, называемой A в Main. Акт отправки замыкания ()->sum(A) в worker 2 приводит к определению Main.A на 2. Main.A продолжает существовать на worker 2 даже после возвращения вызова remotecall_fetch. Удалённые вызовы с вложенными глобальными ссылками (только в модуле Main ) управляют глобальными переменными следующим образом:

• Новые глобальные привязки создаются на целевых worker, если на них есть ссылки в рамках удалённого вызова.

• Глобальные константы объявляются как константы и на удалённых узлах.

• Глобальные переменные отправляются на целевой worker только в контексте удалённого вызова, и только если их значение изменилось. Кроме того, кластер не синхронизирует глобальные привязки между узлами. Например:

  A = rand(10,10)
  remotecall_fetch(()->sum(A), 2) # worker 2
  A = rand(10,10)
  remotecall_fetch(()->sum(A), 3) # worker 3
  A = nothing

  Выполнение указанного фрагмента кода приводит к тому, что Main.A на worker 2 имеет другое значение, чем Main.A на worker 3, в то время как значение Main.A на узле 1 установлено в nothing.

Как вы могли понять, хотя память, связанная с глобальными переменными, может быть освобождена при их повторной присвоении на главном процессе, аналогичные действия не выполняются на worker, так как привязки остаются действительными. clear! может быть использован для ручного повторного назначения определённых глобальных переменных на удалённых узлах в nothing после того, как они больше не требуются. Это позволит освободить память, связанную с ними, в ходе обычного цикла сбора мусора.

Поэтому программы должны быть осторожны при использовании ссылок на глобальные переменные в удалённых вызовах. На самом деле предпочтительнее вообще избегать их использования, если это возможно. Если вам необходимо ссылаться на глобальные переменные, рассмотрите использование let блоков для локализации глобальных переменных.

Например:

julia> A = rand(10,10);

julia> remotecall_fetch(()->A, 2);

julia> B = rand(10,10);

julia> let B = B
           remotecall_fetch(()->B, 2)
       end;

julia> @fetchfrom 2 InteractiveUtils.varinfo()
name           size summary
––––––––– ––––––––– ––––––––––––––––––––––
A         800 bytes 10×10 Array{Float64,2}
Base                Module
Core                Module
Main                Module

Как видно, глобальная переменная A определена на worker 2, но B захвачена как локальная переменная, и поэтому привязки к B не существует на worker 2.

Параллельные циклы и Map

К счастью, многие полезные параллельные вычисления не требуют перемещения данных. Типичный пример — моделирование Монте-Карло, где несколько процессов могут обрабатывать независимые испытания моделирования одновременно. Мы можем использовать @spawnat для подбрасывания монет на двух процессах. Сначала напишем следующую функцию в count_heads.jl:

function count_heads(n)
    c::Int = 0
    for i = 1:n
        c += rand(Bool)
    end
    c
end

Функция count_heads просто складывает n случайные биты. Вот как мы можем выполнить несколько испытаний на двух машинах и сложить результаты:

julia> @everywhere include_string(Main, $(read("count_heads.jl", String)), "count_heads.jl")

julia> a = @spawnat :any count_heads(100000000)
Future(2, 1, 6, nothing)

julia> b = @spawnat :any count_heads(100000000)
Future(3, 1, 7, nothing)

julia> fetch(a)+fetch(b)
100001564

Этот пример демонстрирует мощный и часто используемый шаблон параллельного программирования. Много итераций выполняются независимо на нескольких процессах, а затем их результаты объединяются с помощью какой-либо функции. Этот процесс объединения называется редукцией, так как он обычно выполняет сокращение тензорного ранга: вектор чисел сводится к одному числу, матрица сводится к одной строке или столбцу и т. д. В коде это обычно выглядит как шаблон x = f(x,v[i]), где x — это аккумулирующая переменная, f — функция редукции, а v[i] — это элементы, подлежащие редукции. Желательно, чтобы f была ассоциативной, чтобы порядок выполнения операций не имел значения.

Заметьте, что наше использование этого шаблона с count_heads может быть обобщено. Мы использовали две явные инструкции @spawnat, что ограничивает параллелизм двумя процессами. Для выполнения на любом количестве процессов мы можем использовать параллельный цикл for, работающий в распределённой памяти, который можно записать в Julia с помощью @distributed вот так:

nheads = @distributed (+) for i = 1:200000000
    Int(rand(Bool))
end

Эта конструкция реализует шаблон назначения итераций нескольким процессам и их объединения с помощью указанной редукции (в данном случае (+)). Результат каждой итерации принимается как значение последнего выражения внутри цикла. Всё выражение параллельного цикла само вычисляется до окончательного ответа.

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

Например, следующий код не будет работать должным образом:

a = zeros(100000)
@distributed for i = 1:100000
    a[i] = i
end

Этот код не инициализирует все a, так как каждый процесс будет иметь отдельную копию. Параллельные циклы for, подобные этому, следует избегать. К счастью, Shared Arrays можно использовать, чтобы обойти это ограничение:

using SharedArrays

a = SharedArray{Float64}(10)
@distributed for i = 1:10
    a[i] = i
end

Использование «внешних» переменных в параллельных циклах вполне допустимо, если переменные являются только для чтения:

a = randn(1000)
@distributed (+) for i = 1:100000
    f(a[rand(1:end)])
end

Здесь каждая итерация применяет f к случайно выбранному образцу из вектора a, общий для всех процессов.

Как вы можете видеть, оператор сокращения можно опустить, если он не нужен. В этом случае цикл выполняется асинхронно, т.е. он порождает независимые задачи на всех доступных рабочих процессах и возвращает массив Future немедленно, не дожидаясь завершения. Вызывающий код может дождаться завершения Future в более поздний момент, вызвав fetch на них, или дождаться завершения в конце цикла, добавив префикс @sync, как @sync @distributed for.

В некоторых случаях оператор сокращения не нужен, и мы просто хотим применить функцию ко всем целым числам в некотором диапазоне (или, более обобщенно, ко всем элементам в некотором наборе). Это ещё одна полезная операция, называемая параллельным отображением, реализованная в Julia как функция pmap. Например, мы можем вычислить сингулярные значения нескольких больших случайных матриц параллельно следующим образом:

julia> M = Matrix{Float64}[rand(1000,1000) for i = 1:10];

julia> pmap(svdvals, M);

Функция Julia pmap предназначена для случаев, когда каждый вызов функции выполняет большой объём работы. В отличие от этого, @distributed for может обрабатывать ситуации, когда каждая итерация очень мала, например, простое сложение двух чисел. Только рабочие процессы используются как функцией pmap, так и @distributed for для параллельного вычисления. В случае @distributed for, окончательное сокращение выполняется на вызывающем процессе.

Удаленные ссылки и абстрактные каналы

Удаленные ссылки всегда ссылаются на реализацию AbstractChannel.

Для реализации put!, take!, fetch, isready и wait требуется конкретная реализация AbstractChannel (например, Channel). Удаленный объект, на который ссылается Future, хранится в Channel{Any}(1), то есть Channel размера 1, способном хранить объекты типа Any.

RemoteChannel, которая переписывается, может указывать на каналы любого типа и размера, или на любую другую реализацию AbstractChannel.

Конструктор RemoteChannel(f::Function, pid)() позволяет нам создавать ссылки на каналы, содержащие более одного значения определённого типа. f — это функция, выполняемая на pid, и она должна возвращать AbstractChannel.

Например, RemoteChannel(()->Channel{Int}(10), pid), вернёт ссылку на канал типа Int и размера 10. Канал существует на рабочем процессе pid.

Методы put!, take!, fetch, isready и wait для RemoteChannel делегируются на подлежащий хранилищу удалённый процесс.

RemoteChannel может таким образом использоваться для ссылки на объекты AbstractChannel пользователя. Простым примером этого является следующий DictChannel, который использует словарь в качестве удалённого хранилища:

julia> struct DictChannel{T} <: AbstractChannel{T}
           d::Dict
           cond_take::Threads.Condition    # waiting for data to become available
           DictChannel{T}() where {T} = new(Dict(), Threads.Condition())
           DictChannel() = DictChannel{Any}()
       end

julia> begin
       function Base.put!(D::DictChannel, k, v)
           @lock D.cond_take begin
               D.d[k] = v
               notify(D.cond_take)
           end
           return D
       end
       function Base.take!(D::DictChannel, k)
           @lock D.cond_take begin
               v = fetch(D, k)
               delete!(D.d, k)
               return v
           end
       end
       Base.isready(D::DictChannel) = @lock D.cond_take !isempty(D.d)
       Base.isready(D::DictChannel, k) = @lock D.cond_take haskey(D.d, k)
       function Base.fetch(D::DictChannel, k)
           @lock D.cond_take begin
               wait(D, k)
               return D.d[k]
           end
       end
       function Base.wait(D::DictChannel, k)
           @lock D.cond_take begin
               while !isready(D, k)
                   wait(D.cond_take)
               end
           end
       end
       end;

julia> d = DictChannel();

julia> isready(d)
false

julia> put!(d, :k, :v);

julia> isready(d, :k)
true

julia> fetch(d, :k)
:v

julia> wait(d, :k)

julia> take!(d, :k)
:v

julia> isready(d, :k)
false

Каналы и RemoteChannels

• Канал Channel локален для процесса. Рабочий процесс 2 не может напрямую обратиться к Channel на рабочем процессе 3 и наоборот. Однако, RemoteChannel может отправлять и получать значения между рабочими процессами.
• Можно рассматривать RemoteChannel как указатель на Channel.
• Идентификатор процесса, pid, связанный с RemoteChannel, определяет процесс, в котором находится базовая структура данных, то есть базовый Channel.
• Любой процесс с ссылкой на RemoteChannel может отправлять и получать элементы из канала. Данные автоматически отправляются (или извлекаются) в процесс, с которым связан RemoteChannel.
• Сериализация Channel также сериализует любые данные, присутствующие в канале. Десериализация, таким образом, фактически создаёт копию исходного объекта.
• С другой стороны, сериализация RemoteChannel включает только сериализацию идентификатора, который определяет расположение и экземпляр Channel, на который ссылается указатель. Таким образом, десериализованный объект RemoteChannel (на любом рабочем процессе) также указывает на то же самое базовый хранилище, что и оригинал.

Пример каналов из вышеизложенного можно модифицировать для межпроцессного взаимодействия, как показано ниже.

Мы запускаем 4 рабочих процесса для обработки единственного jobs удалённого канала. Задачи, идентифицированные по id (job_id), записываются в канал. Каждая удалённо выполняемая задача в этом симуляции считывает job_id, ждёт случайное количество времени и записывает назад кортеж из job_id, времени выполнения и своего собственного pid в канал результатов. Наконец, все results выводятся на главном процессе.

julia> addprocs(4); # add worker processes

julia> const jobs = RemoteChannel(()->Channel{Int}(32));

julia> const results = RemoteChannel(()->Channel{Tuple}(32));

julia> @everywhere function do_work(jobs, results) # define work function everywhere
           while true
               job_id = take!(jobs)
               exec_time = rand()
               sleep(exec_time) # simulates elapsed time doing actual work
               put!(results, (job_id, exec_time, myid()))
           end
       end

julia> function make_jobs(n)
           for i in 1:n
               put!(jobs, i)
           end
       end;

julia> n = 12;

julia> errormonitor(@async make_jobs(n)); # feed the jobs channel with "n" jobs

julia> for p in workers() # start tasks on the workers to process requests in parallel
           remote_do(do_work, p, jobs, results)
       end

julia> @elapsed while n > 0 # print out results
           job_id, exec_time, where = take!(results)
           println("$job_id finished in $(round(exec_time; digits=2)) seconds on worker $where")
           global n = n - 1
       end
1 finished in 0.18 seconds on worker 4
2 finished in 0.26 seconds on worker 5
6 finished in 0.12 seconds on worker 4
7 finished in 0.18 seconds on worker 4
5 finished in 0.35 seconds on worker 5
4 finished in 0.68 seconds on worker 2
3 finished in 0.73 seconds on worker 3
11 finished in 0.01 seconds on worker 3
12 finished in 0.02 seconds on worker 3
9 finished in 0.26 seconds on worker 5
8 finished in 0.57 seconds on worker 4
10 finished in 0.58 seconds on worker 2
0.055971741

Удаленные ссылки и распределённое удаление мусора

Объекты, на которые ссылаются удаленные ссылки, могут быть освобождены только тогда, когда все удерживающие ссылки в кластере будут удалены.

Узел, на котором хранится значение, отслеживает, у каких рабочих процессов есть ссылка на него. Каждый раз, когда RemoteChannel или (неполученный) Future сериализуется в рабочий процесс, узел, на который указывает ссылка, уведомляется. И каждый раз, когда RemoteChannel или (неполученный) Future удаляется локально, узел, владеющий значением, снова уведомляется. Это реализовано в внутренней кластерной сериализации. Удаленные ссылки действительны только в контексте работающего кластера. Сериализация и десериализация ссылок на обычные IO объекты не поддерживается.

Уведомления выполняются путём отправки сообщений «отслеживания» — сообщение «добавить ссылку» при сериализации ссылки в другой процесс и сообщение «удалить ссылку» при локальном удалении ссылки.

Поскольку Future записываются один раз и кешируются локально, действие fetch Future также обновляет информацию об отслеживании ссылок на узле, владеющем значением.

Узел, который владеет значением, освобождает его, когда все ссылки на него будут удалены.

С Future, сериализация уже полученного Future в другой узел также отправляет значение, поскольку исходное удалённое хранилище может к этому времени собрать значение.

Важно отметить, что когда объект удаляется локально, зависит от размера объекта и текущей нагрузки памяти в системе.

В случае удалённых ссылок размер локального объекта ссылки довольно мал, в то время как значение, хранящееся на удалённом узле, может быть довольно большим. Поскольку локальный объект может не быть удалён немедленно, рекомендуется явно вызывать finalize на локальных экземплярах RemoteChannel, или на неполученных Future. Поскольку вызов fetch на Future также удаляет его ссылку из удалённого хранилища, это не требуется для полученных Future. Явное вызов finalize приводит к немедленному отправлению сообщения удалённому узлу, чтобы он удалил свою ссылку на значение.

После финализации ссылка становится недействительной и не может быть использована в дальнейших вызовах.

Локальные вызовы

Данные необходимо скопировать на удалённый узел для выполнения. Это происходит как для удалённых вызовов, так и при сохранении данных в RemoteChannel / Future на другом узле. Как ожидается, это приводит к созданию копии сериализованных объектов на удалённом узле. Однако, когда целевой узел — это локальный узел, т.е. идентификатор вызывающего процесса совпадает с идентификатором удалённого узла, вызов выполняется как локальный вызов. Обычно (но не всегда) он выполняется в другой задаче — но нет сериализации/десериализации данных. Следовательно, вызов ссылается на те же экземпляры объектов, что и переданы — копии не создаются. Это поведение показано ниже:

julia> using Distributed;

julia> rc = RemoteChannel(()->Channel(3));   # RemoteChannel created on local node

julia> v = [0];

julia> for i in 1:3
           v[1] = i                          # Reusing `v`
           put!(rc, v)
       end;

julia> result = [take!(rc) for _ in 1:3];

julia> println(result);
Array{Int64,1}[[3], [3], [3]]

julia> println("Num Unique objects : ", length(unique(map(objectid, result))));
Num Unique objects : 1

julia> addprocs(1);

julia> rc = RemoteChannel(()->Channel(3), workers()[1]);   # RemoteChannel created on remote node

julia> v = [0];

julia> for i in 1:3
           v[1] = i
           put!(rc, v)
       end;

julia> result = [take!(rc) for _ in 1:3];

julia> println(result);
Array{Int64,1}[[1], [2], [3]]

julia> println("Num Unique objects : ", length(unique(map(objectid, result))));
Num Unique objects : 3

Как видно, put! на локальном RemoteChannel с тем же объектом v модифицированным между вызовами, приводит к сохранению того же единственного экземпляра объекта. В отличие от копий v при создании, когда узел, владеющий rc находится на другом узле.

Следует отметить, что это, как правило, не проблема. Это необходимо учитывать только в том случае, если объект хранится локально и модифицируется после вызова. В таких случаях может быть целесообразно сохранить deepcopy объекта.

Это также верно для удаленных вызовов на локальном узле, как показано в следующем примере:

julia> using Distributed; addprocs(1);

julia> v = [0];

julia> v2 = remotecall_fetch(x->(x[1] = 1; x), myid(), v);     # Executed on local node

julia> println("v=$v, v2=$v2, ", v === v2);
v=[1], v2=[1], true

julia> v = [0];

julia> v2 = remotecall_fetch(x->(x[1] = 1; x), workers()[1], v); # Executed on remote node

julia> println("v=$v, v2=$v2, ", v === v2);
v=[0], v2=[1], false

Как видно ещё раз, удалённый вызов на локальный узел ведет себя как прямой вызов. Вызов изменяет локальные объекты, переданные в качестве аргументов. При удалённом вызове он работает с копией аргументов.

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

Общие массивы

Общие массивы используют общую системную память для отображения одного массива на многих процессах. SharedArray — хороший выбор, когда вам нужно, чтобы большой объем данных был совместно доступен двум или более процессам на одном компьютере. Поддержка общих массивов доступна через модуль SharedArrays, который необходимо явно загрузить на все участвующие рабочие процессы.

Дополнительная структура данных предоставляется внешним пакетом DistributedArrays.jl в виде DArray. Хотя есть некоторые сходства с SharedArray, поведение DArray довольно отличается. В SharedArray каждый «участвующий» процесс имеет доступ ко всему массиву; в отличие от DArray, каждый процесс имеет локальный доступ только к части данных, и никакие два процесса не разделяют одну и ту же часть.

SharedArray индексация (присваивание и доступ к значениям) работает так же, как и с обычными массивами, и является эффективной, потому что подлежащая память доступна локальному процессу. Поэтому большинство алгоритмов естественным образом работают с SharedArray, хотя и в режиме одного процесса. В тех случаях, когда алгоритм требует Array входных данных, лежащий в основе массив может быть извлечён из SharedArray вызовом sdata. Для других AbstractArray типов sdata просто возвращает сам объект, поэтому sdata безопасно использовать с любым Array объектом.

Конструктор для общего массива имеет вид:

SharedArray{T,N}(dims::NTuple; init=false, pids=Int[])

что создает N-мерный общий массив типа T и размера dims по всем процессам, указанным в pids. В отличие от распределенных массивов, общий массив доступен только из тех рабочих процессов, которые указаны в аргументе pids (и также из создающего процесса, если он находится на том же хосте). Обратите внимание, что в SharedArray поддерживаются только элементы типа isbits.

Если указана функция init с сигнатурой initfn(S::SharedArray), она вызывается на всех участвующих рабочих процессах. Вы можете указать, чтобы каждый рабочий процесс выполнял функцию init на отдельной части массива, тем самым обеспечивая параллелизацию инициализации.

Вот краткий пример:

julia> using Distributed

julia> addprocs(3)
3-element Array{Int64,1}:
 2
 3
 4

julia> @everywhere using SharedArrays

julia> S = SharedArray{Int,2}((3,4), init = S -> S[localindices(S)] = repeat([myid()], length(localindices(S))))
3×4 SharedArray{Int64,2}:
 2  2  3  4
 2  3  3  4
 2  3  4  4

julia> S[3,2] = 7
7

julia> S
3×4 SharedArray{Int64,2}:
 2  2  3  4
 2  3  3  4
 2  7  4  4

SharedArrays.localindices предоставляет непересекающиеся одномерные диапазоны индексов и иногда удобен для разделения задач между процессами. Вы, конечно, можете разделить работу любым способом:

julia> S = SharedArray{Int,2}((3,4), init = S -> S[indexpids(S):length(procs(S)):length(S)] = repeat([myid()], length( indexpids(S):length(procs(S)):length(S))))
3×4 SharedArray{Int64,2}:
 2  2  2  2
 3  3  3  3
 4  4  4  4

Так как все процессы имеют доступ к данным, необходимо быть осторожными, чтобы не создать конфликт. Например:

@sync begin
    for p in procs(S)
        @async begin
            remotecall_wait(fill!, p, S, p)
        end
    end
end

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

В качестве более расширенного и сложного примера рассмотрим запуск следующего "ядра" параллельно:

q[i,j,t+1] = q[i,j,t] + u[i,j,t]

В этом случае, если мы попытаемся разделить работу с использованием одномерного индекса, мы, скорее всего, столкнемся с проблемами: если q[i,j,t] находится в конце блока, назначенного одному рабочему процессу, а q[i,j,t+1] — в начале блока, назначенного другому, очень вероятно, что q[i,j,t] не будет готов в тот момент, когда он нужен для вычисления q[i,j,t+1]. В таких случаях лучше вручную разбить массив на части. Давайте разделим по второму измерению. Определим функцию, которая возвращает индексы (irange, jrange) , назначенные этому рабочему процессу:

julia> @everywhere function myrange(q::SharedArray)
           idx = indexpids(q)
           if idx == 0 # This worker is not assigned a piece
               return 1:0, 1:0
           end
           nchunks = length(procs(q))
           splits = [round(Int, s) for s in range(0, stop=size(q,2), length=nchunks+1)]
           1:size(q,1), splits[idx]+1:splits[idx+1]
       end

Далее определим ядро:

julia> @everywhere function advection_chunk!(q, u, irange, jrange, trange)
           @show (irange, jrange, trange)  # display so we can see what's happening
           for t in trange, j in jrange, i in irange
               q[i,j,t+1] = q[i,j,t] + u[i,j,t]
           end
           q
       end

Также определим удобную оболочку для реализации SharedArray.

julia> @everywhere advection_shared_chunk!(q, u) =
           advection_chunk!(q, u, myrange(q)..., 1:size(q,3)-1)

Теперь давайте сравним три различных версии, одна из которых выполняется в одном процессе:

julia> advection_serial!(q, u) = advection_chunk!(q, u, 1:size(q,1), 1:size(q,2), 1:size(q,3)-1);

другая использует @distributed:

julia> function advection_parallel!(q, u)
           for t = 1:size(q,3)-1
               @sync @distributed for j = 1:size(q,2)
                   for i = 1:size(q,1)
                       q[i,j,t+1]= q[i,j,t] + u[i,j,t]
                   end
               end
           end
           q
       end;

и третья, которая делегирует по частям:

julia> function advection_shared!(q, u)
           @sync begin
               for p in procs(q)
                   @async remotecall_wait(advection_shared_chunk!, p, q, u)
               end
           end
           q
       end;

Если мы создадим SharedArray и измерим время работы этих функций, мы получим следующие результаты (с julia -p 4):

julia> q = SharedArray{Float64,3}((500,500,500));

julia> u = SharedArray{Float64,3}((500,500,500));

Запустите функции один раз, чтобы скомпилировать их и запустить @time во второй раз:

julia> @time advection_serial!(q, u);
(irange,jrange,trange) = (1:500,1:500,1:499)
 830.220 milliseconds (216 allocations: 13820 bytes)

julia> @time advection_parallel!(q, u);
   2.495 seconds      (3999 k allocations: 289 MB, 2.09% gc time)

julia> @time advection_shared!(q,u);
        From worker 2:       (irange,jrange,trange) = (1:500,1:125,1:499)
        From worker 4:       (irange,jrange,trange) = (1:500,251:375,1:499)
        From worker 3:       (irange,jrange,trange) = (1:500,126:250,1:499)
        From worker 5:       (irange,jrange,trange) = (1:500,376:500,1:499)
 238.119 milliseconds (2264 allocations: 169 KB)

Главным преимуществом advection_shared! является минимизация трафика между рабочими процессами, позволяя каждому выполнять вычисления в течение длительного времени на назначенном участке.

Общие массивы и распределённое управление сборкой мусора

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

Управляющие кластерами

Запуск, управление и организация сети процессов Julia в логическом кластере выполняется через управляющие кластерами. ClusterManager отвечает за

• запуск рабочих процессов в среде кластера
• управление событиями на протяжении всего жизненного цикла каждого рабочего процесса
• необязательно, предоставление транспорта данных

Кластер Julia имеет следующие характеристики:

• Изначальный процесс Julia, также называемый master, является специальным и имеет id равный 1.
• Только процесс master может добавлять или удалять рабочие процессы.
• Все процессы могут напрямую общаться друг с другом.

Подключение между рабочими процессами (с использованием встроенного транспорта TCP/IP) устанавливается следующим образом:

• addprocs вызывается в главном процессе с объектом ClusterManager.
• addprocs вызывает соответствующий метод launch, который порождает необходимое количество рабочих процессов на соответствующих машинах.
• Каждый рабочий процесс начинает прослушивание на свободном порту и записывает информацию о своём хосте и порте в stdout.
• Управляющий кластером перехватывает stdout каждого рабочего процесса и делает его доступным для главного процесса.
• Главный процесс анализирует эту информацию и устанавливает соединения TCP/IP с каждым рабочим процессом.
• Каждый рабочий процесс также уведомляется об остальных рабочих процессах в кластере.
• Каждый рабочий процесс подключается ко всем рабочим процессам, у которых id меньше, чем у собственного рабочего процесса id.
• Таким образом, устанавливается сеть типа "mesh", в которой каждый рабочий процесс напрямую подключен к каждому другому.

В то время как стандартный уровень транспорта использует обычные TCPSocket, для кластера Julia возможно предоставление собственного транспорта.

Julia предоставляет два встроенных управляющих кластерами:

• LocalManager, используемый при вызове addprocs() или addprocs(np::Integer)
• SSHManager, используемый при вызове addprocs(hostnames::Array) со списком имён хостов

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

Таким образом, минимальный управляющий кластерами должен:

• быть подтипом абстрактного ClusterManager
• реализовывать launch, метод, отвечающий за запуск новых рабочих процессов
• реализовывать manage, который вызывается при различных событиях в течение жизненного цикла рабочего процесса (например, для отправки сигнала прерывания)

addprocs(manager::FooManager) требует, чтобы FooManager реализовали:

function launch(manager::FooManager, params::Dict, launched::Array, c::Condition)
    [...]
end

function manage(manager::FooManager, id::Integer, config::WorkerConfig, op::Symbol)
    [...]
end

В качестве примера давайте рассмотрим, как реализуется LocalManager, управляющий кластером, ответственный за запуск рабочих процессов на одном хосте:

struct LocalManager <: ClusterManager
    np::Integer
end

function launch(manager::LocalManager, params::Dict, launched::Array, c::Condition)
    [...]
end

function manage(manager::LocalManager, id::Integer, config::WorkerConfig, op::Symbol)
    [...]
end

Метод launch принимает следующие аргументы:

• manager::ClusterManager: менеджер кластера, с которым вызывается addprocs
• params::Dict: все именованные аргументы, переданные в addprocs
• launched::Array: массив, в который нужно добавить один или несколько объектов WorkerConfig
• c::Condition: переменная состояния, которая уведомляется по мере запуска рабочих процессов

Метод launch вызывается асинхронно в отдельной задаче. Завершение этой задачи сигнализирует о том, что все запрошенные рабочие процессы были запущены. Таким образом, функция launch ДОЛЖНА завершиться сразу же после запуска всех запрошенных рабочих процессов.

Недавно запущенные рабочие процессы подключаются друг к другу и к мастер-процессу по схеме «все со всеми». Указание командной строки --worker[=<cookie>] приводит к тому, что запущенные процессы инициализируются как рабочие процессы, а соединения устанавливаются через сокеты TCP/IP.

Все рабочие процессы в кластере используют тот же cookie, что и мастер. Если cookie не указан, т.е. с опцией --worker, рабочий процесс пытается прочитать его из стандартного ввода. LocalManager и SSHManager передают cookie в недавно запущенные рабочие процессы через их стандартный ввод.

По умолчанию рабочий процесс будет прослушивать свободный порт по адресу, возвращаемому вызовом getipaddr(). Указать конкретный адрес для прослушивания можно с помощью необязательного аргумента --bind-to bind_addr[:port]. Это полезно для систем с несколькими сетевыми интерфейсами.

В качестве примера не-TCP/IP транспорта, реализация может выбрать использование MPI, в этом случае --worker НЕ должно быть указано. Вместо этого, недавно запущенные рабочие процессы должны вызвать init_worker(cookie) перед использованием любых параллельных конструкций.

Для каждого запущенного рабочего процесса метод launch должен добавить объект WorkerConfig (с соответствующими инициализированными полями) в launched

mutable struct WorkerConfig
    # Common fields relevant to all cluster managers
    io::Union{IO, Nothing}
    host::Union{AbstractString, Nothing}
    port::Union{Integer, Nothing}

    # Used when launching additional workers at a host
    count::Union{Int, Symbol, Nothing}
    exename::Union{AbstractString, Cmd, Nothing}
    exeflags::Union{Cmd, Nothing}

    # External cluster managers can use this to store information at a per-worker level
    # Can be a dict if multiple fields need to be stored.
    userdata::Any

    # SSHManager / SSH tunnel connections to workers
    tunnel::Union{Bool, Nothing}
    bind_addr::Union{AbstractString, Nothing}
    sshflags::Union{Cmd, Nothing}
    max_parallel::Union{Integer, Nothing}

    # Used by Local/SSH managers
    connect_at::Any

    [...]
end

Большинство полей в WorkerConfig используются встроенными менеджерами. Пользовательские менеджеры кластера обычно указывают только io или host / port.

• Если io указан, он используется для чтения информации о хосте/порту. Рабочий процесс Julia выводит свой адрес и порт при запуске. Это позволяет рабочим процессам Julia прослушивать любой свободный порт, вместо того, чтобы вручную настраивать порты рабочих процессов.

• Если io не указан, используются host и port для подключения.

• count, exename и exeflags важны для запуска дополнительных рабочих процессов из рабочего процесса. Например, менеджер кластера может запускать по одному рабочему процессу на узел и использовать его для запуска дополнительных рабочих процессов.

  • count со значением в виде целого числа n запустит общее количество n рабочих процессов.
  • count со значением :auto запустит столько рабочих процессов, сколько потоков CPU (логических ядер) на этой машине.
  • exename — имя исполняемого файла julia, включая полный путь.
  • exeflags должен содержать требуемые аргументы командной строки для новых рабочих процессов.

• tunnel, bind_addr, sshflags и max_parallel используются при необходимости SSH-туннеля для подключения к рабочим процессам из мастер-процесса.

• userdata предоставлен пользовательским менеджерам кластера для хранения собственной информации о рабочих процессах.

manage(manager::FooManager, id::Integer, config::WorkerConfig, op::Symbol) вызывается в разные моменты жизни рабочего процесса с соответствующими значениями op.

• со значениями :register/:deregister при добавлении/удалении рабочего процесса из пула рабочих процессов Julia.
• со значением :interrupt при вызове interrupt(workers). ClusterManager должен отправить соответствующему рабочему процессу сигнал прерывания.
• со значением :finalize для целей очистки.

Менеджеры кластера с пользовательскими средствами транспорта

Замена стандартных соединений через TCP/IP-сокеты на пользовательский уровень транспорта немного сложнее. Каждый процесс Julia имеет столько задач связи, сколько рабочих процессов с ним подключено. Например, рассмотрим кластер Julia из 32 процессов в сети с полносвязным соединением:

• Каждый процесс Julia имеет 31 задачу связи.
• Каждая задача обрабатывает все входящие сообщения от одного удаленного рабочего процесса в цикле обработки сообщений.
• Цикл обработки сообщений ожидает объекта IO (например, TCPSocket в реализации по умолчанию), читает всё сообщение, обрабатывает его и ожидает следующего.
• Отправка сообщений в процесс выполняется непосредственно из любой задачи Julia — не только задач связи — опять же, через соответствующий объект IO.

Для замены стандартного транспорта новая реализация должна настроить соединения с удаленными рабочими процессами и предоставить соответствующие объекты IO, на которых циклы обработки сообщений могут ожидать. Менеджер-специфические обратные вызовы, которые необходимо реализовать, это:

connect(manager::FooManager, pid::Integer, config::WorkerConfig)
kill(manager::FooManager, pid::Int, config::WorkerConfig)

Реализация по умолчанию (которая использует сокеты TCP/IP) реализована как connect(manager::ClusterManager, pid::Integer, config::WorkerConfig).

connect должен вернуть пару объектов IO — один для чтения данных, отправленных рабочим процессом pid, и другой для записи данных, которые необходимо отправить рабочему процессу pid. Пользовательские менеджеры кластера могут использовать в памяти BufferStream как коммуникационную трубу для передачи данных между пользовательским, возможно не-IO транспортом и встроенной параллельной инфраструктурой Julia.

BufferStream — это в памяти IOBuffer, который ведет себя как IO — это поток, который можно обрабатывать асинхронно.

Папка clustermanager/0mq в репозитории примеров содержит пример использования ZeroMQ для подключения рабочих процессов Julia в звездообразной топологии с 0MQ-брокером посередине. Примечание: процессы Julia всё ещё логически соединены друг с другом — любой рабочий процесс может отправлять сообщения любому другому рабочему процессу напрямую, без какого-либо понимания использования 0MQ как уровня транспорта.

При использовании пользовательских средств транспорта:

• Рабочие процессы Julia НЕ должны запускаться с --worker. Запуск с --worker приведет к тому, что недавно запущенные рабочие процессы будут использовать реализацию транспорта TCP/IP сокетов по умолчанию.
• Для каждого входящего логического соединения с рабочим процессом должен быть вызван Base.process_messages(rd::IO, wr::IO)(). Это запускает новую задачу, которая обрабатывает чтение и запись сообщений из/в рабочий процесс, представленный объектами IO.
• init_worker(cookie, manager::FooManager) обязательно должен быть вызван в рамках инициализации процесса рабочего процесса.
• Поле connect_at::Any в WorkerConfig может быть установлено менеджером кластера при вызове launch. Значение этого поля передается во всех обратных вызовах connect. Обычно оно содержит информацию о том, как подключиться к рабочему процессу. Например, транспорт TCP/IP сокетов использует это поле для указания кортежа (host, port) для подключения к рабочему процессу.

kill(manager, pid, config) вызывается для удаления рабочего процесса из кластера. В мастер-процессе соответствующие объекты IO должны быть закрыты реализацией для обеспечения надлежащей очистки. Реализация по умолчанию просто выполняет вызов exit() в указанном удалённом рабочем процессе.

Папка примеров clustermanager/simple — это пример простой реализации с использованием сокетов доменной системы Unix для настройки кластера.

Требования к сети для LocalManager и SSHManager

Кластеры Julia предназначены для выполнения в уже защищённых средах на инфраструктуре, такой как локальные ноутбуки, ведомственные кластеры или даже облако. Этот раздел описывает требования к сетевой безопасности для встроенных LocalManager и SSHManager.

• Главный процесс не прослушивает ни один порт. Он только подключается к рабочим процессам.

• Каждый рабочий процесс связывается только с одним из локальных интерфейсов и прослушивает порт со случайным номером, назначенным ОС.

• LocalManager, используемый addprocs(N), по умолчанию связывается только с локальным интерфейсом. Это означает, что рабочие процессы, запущенные позже на удаленных хостах (или кем-либо с недобросовестными намерениями), не могут подключиться к кластеру. Попытка addprocs(4) с последующим addprocs(["remote_host"]) завершится ошибкой. Некоторые пользователи могут нуждаться в создании кластера, включающего их локальную систему и несколько удаленных систем. Это можно сделать, явно указав LocalManager на привязку к внешнему сетевому интерфейсу через ключевое слово restrict: addprocs(4; restrict=false).

• SSHManager, используемый addprocs(list_of_remote_hosts), запускает рабочие процессы на удаленных хостах через SSH. По умолчанию SSH используется только для запуска рабочих процессов Julia. Последующие подключения мастер-рабочий и рабочий-рабочий используют обычные, незашифрованные TCP/IP-сокеты. На удаленных хостах должна быть включена возможность входа без пароля. Дополнительные флаги SSH или учетные данные могут быть указаны через ключевое слово sshflags.

• addprocs(list_of_remote_hosts; tunnel=true, sshflags=<ssh keys and other flags>) полезно, когда необходимо использовать SSH-соединения и для мастер-рабочих подключений. Типичный сценарий для этого — локальный ноутбук с запуском Julia REPL (т.е., мастер) и остальная часть кластера в облаке, например, на Amazon EC2. В этом случае необходимо открыть только порт 22 в удаленном кластере, а также обеспечить аутентификацию SSH-клиента через инфраструктуру открытых ключей (PKI). Учетные данные аутентификации могут быть предоставлены через sshflags, например sshflags=`-i <keyfile>`.

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

  Защита и шифрование всего трафика между рабочими процессами (через SSH) или шифрование отдельных сообщений можно выполнить с помощью настраиваемого ClusterManager.

• Если вы укажете multiplex=true в качестве опции для addprocs, будет использоваться SSH-мультиплексирование для создания туннеля между мастером и рабочими процессами. Если вы настроите SSH-мультиплексирование самостоятельно и подключение уже установлено, SSH-мультиплексирование используется независимо от опции multiplex. При включенном мультиплексировании перенаправление устанавливается с использованием существующего соединения (опция -O forward в ssh). Это полезно, если ваши серверы требуют аутентификации по паролю; вы можете обойти аутентификацию в Julia, войдя на сервер до addprocs. Сокет управления будет расположен по адресу ~/.ssh/julia-%r@%h:%p во время сессии, если не используется существующее подключение мультиплексирования. Обратите внимание, что пропускная способность может быть ограничена, если вы создадите несколько процессов на узле и включите мультиплексирование, поскольку в этом случае процессы используют одно общее TCP-соединение мультиплексирования.

Хэш-ключ кластера

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

• cluster_cookie() возвращает хэш-ключ, а cluster_cookie(cookie)() устанавливает его и возвращает новый хэш-ключ.
• Все подключения аутентифицируются с обеих сторон, чтобы гарантировать, что к другим рабочим процессам могут подключаться только рабочие процессы, запущенные мастером.
• Хэш-ключ может быть передан рабочим процессам при запуске через аргумент --worker=<cookie>. Если аргумент --worker указан без хэш-ключа, рабочий процесс пытается прочитать хэш-ключ из стандартного ввода (stdin). Сокет stdin закрывается сразу после получения хэш-ключа.
• Менеджеры кластера могут получить хэш-ключ на главном сервере, вызвав cluster_cookie(). Менеджеры кластера, не использующие стандартный транспорт TCP/IP (и, следовательно, не указывающие --worker), должны вызвать init_worker(cookie, manager) с тем же хэш-ключом, что и на главном сервере.

Обратите внимание, что среды, требующие более высокого уровня безопасности, могут реализовать это с помощью настраиваемого ClusterManager. Например, хэш-ключи могут быть предварительно объявлены и, следовательно, не указываться в качестве аргумента запуска.

Определение топологии сети (экспериментальная версия)

Ключевое слово topology, переданное в addprocs, используется для определения способа подключения рабочих процессов друг к другу:

• :all_to_all, по умолчанию: все рабочие процессы подключены друг к другу.
• :master_worker: только процесс драйвера, т.е. pid 1, имеет подключения к рабочим процессам.
• :custom: метод launch менеджера кластера определяет топологию подключения через поля ident и connect_idents в WorkerConfig. Рабочий процесс с идентификатором, предоставленным менеджером кластера, ident будет подключаться ко всем рабочим процессам, указанным в connect_idents.

Ключевое слово lazy=true|false влияет только на опцию topology :all_to_all. Если true, кластер запускается с мастером, подключенным ко всем рабочим процессам. Конкретные подключения между рабочими процессами устанавливаются при первом удаленном вызове между двумя рабочими процессами. Это помогает сократить начальные ресурсы, выделенные для внутрикластерного взаимодействия. Подключения устанавливаются в зависимости от текущих требований программы параллельной обработки. Значение по умолчанию для lazy составляет true.

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

Заслуживающие внимания внешние пакеты

Помимо параллелизма Julia, существует множество внешних пакетов, которые следует упомянуть. Например, MPI.jl — это оболочка Julia для протокола MPI, Dagger.jl предоставляет функциональность, аналогичную Python's Dask, а DistributedArrays.jl предоставляет операции над массивами, распределенные по рабочим процессам, как описано выше.

Необходимо упомянуть экосистему GPU-программирования Julia, которая включает:

1. CUDA.jl оборачивает различные библиотеки CUDA и поддерживает компиляцию ядер Julia для графических процессоров Nvidia.

2. oneAPI.jl оборачивает унифицированную модель программирования oneAPI и поддерживает выполнение ядер Julia на поддерживаемых ускорителях. В настоящее время поддерживается только Linux.

3. AMDGPU.jl оборачивает библиотеки AMD ROCm и поддерживает компиляцию ядер Julia для графических процессоров AMD. В настоящее время поддерживается только Linux.

4. Библиотеки высокого уровня, такие как KernelAbstractions.jl, Tullio.jl и ArrayFire.jl.

В следующем примере мы будем использовать как DistributedArrays.jl, так и CUDA.jl, чтобы распределить массив по нескольким процессам, сначала преобразовав его с помощью distribute() и CuArray().

Помните, при импорте DistributedArrays.jl для импорта его во все процессы, используйте @everywhere

$ ./julia -p 4

julia> addprocs()

julia> @everywhere using DistributedArrays

julia> using CUDA

julia> B = ones(10_000) ./ 2;

julia> A = ones(10_000) .* π;

julia> C = 2 .* A ./ B;

julia> all(C .≈ 4*π)
true

julia> typeof(C)
Array{Float64,1}

julia> dB = distribute(B);

julia> dA = distribute(A);

julia> dC = 2 .* dA ./ dB;

julia> all(dC .≈ 4*π)
true

julia> typeof(dC)
DistributedArrays.DArray{Float64,1,Array{Float64,1}}

julia> cuB = CuArray(B);

julia> cuA = CuArray(A);

julia> cuC = 2 .* cuA ./ cuB;

julia> all(cuC .≈ 4*π);
true

julia> typeof(cuC)
CuArray{Float64,1}

В следующем примере мы будем использовать как DistributedArrays.jl, так и CUDA.jl, чтобы распределить массив по нескольким процессам и вызвать на нем универсальную функцию.

function power_method(M, v)
    for i in 1:100
        v = M*v
        v /= norm(v)
    end

    return v, norm(M*v) / norm(v)  # or  (M*v) ./ v
end

power_method многократно создает новый вектор и нормирует его. Мы не указали никакой подписи типа в объявлении функции, давайте посмотрим, будет ли она работать с вышеупомянутыми типами данных:

julia> M = [2. 1; 1 1];

julia> v = rand(2)
2-element Array{Float64,1}:
0.40395
0.445877

julia> power_method(M,v)
([0.850651, 0.525731], 2.618033988749895)

julia> cuM = CuArray(M);

julia> cuv = CuArray(v);

julia> curesult = power_method(cuM, cuv);

julia> typeof(curesult)
CuArray{Float64,1}

julia> dM = distribute(M);

julia> dv = distribute(v);

julia> dC = power_method(dM, dv);

julia> typeof(dC)
Tuple{DistributedArrays.DArray{Float64,1,Array{Float64,1}},Float64}

Чтобы завершить это краткое знакомство с внешними пакетами, можно рассмотреть MPI.jl, оболочку Julia протокола MPI. Поскольку рассмотрение каждой внутренней функции займет слишком много времени, лучше просто оценить подход, используемый для реализации протокола.

Рассмотрим этот пример скрипта, который просто вызывает каждый подпроцесс, инициализирует его ранг, а когда достигает главного процесса, суммирует ранги.

import MPI

MPI.Init()

comm = MPI.COMM_WORLD
MPI.Barrier(comm)

root = 0
r = MPI.Comm_rank(comm)

sr = MPI.Reduce(r, MPI.SUM, root, comm)

if(MPI.Comm_rank(comm) == root)
   @printf("sum of ranks: %s\n", sr)
end

MPI.Finalize()

mpirun -np 4 ./julia example.jl