Распределённые вычисления

Инструменты для распределённой параллельной обработки.

addprocs(manager::ClusterManager; kwargs...) -> List of process identifiers

Запускает рабочие процессы через указанный менеджер кластера.

Например, кластеры Beowulf поддерживаются через пользовательский менеджер кластера, реализованный в пакете ClusterManagers.jl.

Количество секунд, в течение которого запущенный рабочий процесс ждёт установления соединения с главным процессом, можно указать через переменную JULIA_WORKER_TIMEOUT в среде рабочего процесса. Актуально только при использовании TCP/IP в качестве транспорта.

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

Примеры

# On busy clusters, call `addprocs` asynchronously
t = @async addprocs(...)

# Utilize workers as and when they come online
if nprocs() > 1   # Ensure at least one new worker is available
   ....   # perform distributed execution
end

# Retrieve newly launched worker IDs, or any error messages
if istaskdone(t)   # Check if `addprocs` has completed to ensure `fetch` doesn't block
    if nworkers() == N
        new_pids = fetch(t)
    else
        fetch(t)
    end
end

addprocs(machines; tunnel=false, sshflags=``, max_parallel=10, kwargs...) -> List of process identifiers

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

machines — вектор «спецификаций машины», заданных в виде строк формата [user@]host[:port] [bind_addr[:port]]. user по умолчанию устанавливается для текущего пользователя, а port — для стандартного порта SSH. Если указан [bind_addr[:port]], другие рабочие процессы будут подключаться к этому рабочему процессу по указанному bind_addr и port.

Можно запустить несколько процессов на удалённом узле, используя кортеж в векторе machines или форме (machine_spec, count), где count — количество рабочих процессов, которые необходимо запустить на указанном узле. Передача :auto в качестве количества рабочих процессов запустит столько рабочих процессов, сколько потоков ЦП на удалённом узле.

Примеры:

addprocs([
    "remote1",               # one worker on 'remote1' logging in with the current username
    "user@remote2",          # one worker on 'remote2' logging in with the 'user' username
    "user@remote3:2222",     # specifying SSH port to '2222' for 'remote3'
    ("user@remote4", 4),     # launch 4 workers on 'remote4'
    ("user@remote5", :auto), # launch as many workers as CPU threads on 'remote5'
])

Ключевые аргументы:

• tunnel: если true, для подключения к рабочему процессу из главного процесса будет использоваться SSH-туннелирование. По умолчанию — false.

• multiplex: если true, для SSH-туннелирования используется SSH-мультиплексирование. По умолчанию — false.

• ssh: имя или путь к исполняемому файлу SSH-клиента, используемого для запуска рабочих процессов. По умолчанию — "ssh".

• sshflags: указывает дополнительные опции ssh, например, sshflags=`-i /home/foo/bar.pem`

• max_parallel: указывает максимальное количество одновременно подключённых рабочих процессов на узле. По умолчанию — 10.

• shell: указывает тип оболочки, к которой ssh подключается на рабочих процессах.

  • shell=:posix: совместимая с POSIX оболочка Unix/Linux (sh, ksh, bash, dash, zsh и т. д.). По умолчанию.

  • shell=:csh: оболочка Unix C shell (csh, tcsh).

  • shell=:wincmd: Microsoft Windows cmd.exe.

• dir: указывает рабочую директорию на рабочих процессах. По умолчанию — текущая директория узла (как найдена pwd()).

• enable_threaded_blas: если true, BLAS будет выполняться в нескольких потоках в добавленных процессах. По умолчанию — false.

• exename: имя исполняемого файла julia. По умолчанию — "$(Sys.BINDIR)/julia" или "$(Sys.BINDIR)/julia-debug" соответственно. Рекомендуется использовать общую версию Julia на всех удалённых машинах, так как в противном случае могут возникнуть ошибки при сериализации и распределении кода.

• exeflags: дополнительные флаги, передаваемые рабочим процессам.

• topology: Указывает, как рабочие процессы подключаются друг к другу. Передача сообщения между несвязанными рабочими процессами приводит к ошибке.

  • topology=:all_to_all: Все процессы подключены друг к другу. По умолчанию.

  • topology=:master_worker: Только драйверный процесс, т. е. pid 1, подключается к рабочим процессам. Рабочие процессы не подключаются друг к другу.

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

• lazy: Применимо только с topology=:all_to_all. Если true, соединения между рабочими процессами устанавливаются лениво, т. е. они устанавливаются при первом вызове удалённого вызова между рабочими процессами. По умолчанию — true.

• env: предоставляют массив пар строк, таких как env=["JULIA_DEPOT_PATH"=>"/depot"], чтобы запросить установку переменных среды на удалённой машине. По умолчанию автоматически передаётся только переменная среды JULIA_WORKER_TIMEOUT из локальной в удалённую среду.

• cmdline_cookie: передача файла аутентификации через опцию командной строки --worker. Более защищённое поведение (по умолчанию) — передача файла аутентификации через ssh stdio — может зависать с рабочими процессами Windows, использующими более старые (до ConPTY) версии Julia или Windows, в этом случае cmdline_cookie=true предлагает обходной путь.

Переменные среды:

Если главный процесс не может установить соединение с только что запущенным рабочим процессом в течение 60,0 секунд, рабочий процесс рассматривает это как критическую ситуацию и завершается. Этот таймаут можно настроить через переменную среды JULIA_WORKER_TIMEOUT. Значение JULIA_WORKER_TIMEOUT в главном процессе определяет количество секунд, в течение которого запущенный рабочий процесс ждёт установления соединения.

addprocs(np::Integer=Sys.CPU_THREADS; restrict=true, kwargs...) -> List of process identifiers

Запуск np рабочих процессов на локальном узле с использованием встроенного LocalManager.

Локальные рабочие процессы наследуют текущую среду пакетов (т. е. активный проект, LOAD_PATH и DEPOT_PATH) из главного процесса.

Ключевые аргументы:

• restrict::Bool: если true (по умолчанию) привязка ограничена до 127.0.0.1.
• dir, exename, exeflags, env, topology, lazy, enable_threaded_blas: тот же эффект, что и для SSHManager, см. документацию по addprocs(machines::AbstractVector).

nprocs()

Получение количества доступных процессов.

Примеры

julia> nprocs()
3

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

nworkers()

Получение количества доступных рабочих процессов. Это значение на единицу меньше, чем nprocs(). Равно nprocs() если nprocs() == 1.

Примеры

$ julia -p 2

julia> nprocs()
3

julia> nworkers()
2

procs()

Возвращает список идентификаторов всех процессов, включая PID 1 (который не включён в workers()).

Примеры

$ julia -p 2

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

procs(pid::Integer)

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

workers()

Возвращает список идентификаторов всех рабочих процессов.

Примеры

$ julia -p 2

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

rmprocs(pids...; waitfor=typemax(Int))

Удаляет указанные рабочие процессы. Обратите внимание, что только процесс 1 может добавлять или удалять рабочие процессы.

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

• Если не указано, rmprocs будет ждать, пока все запрошенные pids не будут удалены.
• Возникает исключение ErrorException, если все рабочие процессы не могут быть завершены до истечения указанных waitfor секунд.
• Со значением waitfor 0 вызов возвращает немедленно со запланированными для удаления рабочими процессами в другой задаче. Возвращается объект запланированной Task. Пользователь должен вызвать wait для задачи перед вызовом любых других параллельных вызовов.

Примеры

$ julia -p 5

julia> t = rmprocs(2, 3, waitfor=0)
Task (runnable) @0x0000000107c718d0

julia> wait(t)

julia> workers()
3-element Array{Int64,1}:
 4
 5
 6

interrupt(pids::Integer...)

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

interrupt(pids::AbstractVector=workers())

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

myid()

Получение идентификатора текущего процесса.

Примеры

julia> myid()
1

julia> remotecall_fetch(() -> myid(), 4)
4

pmap(f, [::AbstractWorkerPool], c...; distributed=true, batch_size=1, on_error=nothing, retry_delays=[], retry_check=nothing) -> collection

Преобразовать коллекцию c, применяя f к каждому элементу с использованием доступных рабочих процессов и задач.

Для нескольких аргументов коллекции, применяйте f поэлементно.

Обратите внимание, что f должно быть доступно всем рабочим процессам; см. Доступность кода и загрузка пакетов для получения подробностей.

Если пул рабочих процессов не указан, будут использованы все доступные рабочие процессы через CachingPool.

По умолчанию, pmap распределяет вычисления по всем указанным рабочим процессам. Чтобы использовать только локальный процесс и распределить по задачам, укажите distributed=false. Это эквивалентно использованию asyncmap. Например, pmap(f, c; distributed=false) эквивалентно asyncmap(f,c; ntasks=()->nworkers())

pmap также может использовать смесь процессов и задач через аргумент batch_size. Для размеров пакетов больше 1, коллекция обрабатывается в нескольких пакетах, каждый из которых имеет длину batch_size или меньше. Пакетик отправляется как один запрос свободному рабочему процессу, где локальный asyncmap обрабатывает элементы из пакета с использованием нескольких одновременных задач.

Любая ошибка останавливает pmap от обработки оставшейся части коллекции. Чтобы переопределить это поведение, вы можете указать функцию обработки ошибок через аргумент on_error, который принимает один аргумент, т.е. исключение. Функция может остановить обработку, повторно выбросив ошибку, или, чтобы продолжить, вернуть любое значение, которое затем возвращается встраивается в результаты вызывающей функции.

Рассмотрим следующие два примера. Первый возвращает объект исключения встраивая его, второй — 0 вместо любого исключения:

julia> pmap(x->iseven(x) ? error("foo") : x, 1:4; on_error=identity)
4-element Array{Any,1}:
 1
  ErrorException("foo")
 3
  ErrorException("foo")

julia> pmap(x->iseven(x) ? error("foo") : x, 1:4; on_error=ex->0)
4-element Array{Int64,1}:
 1
 0
 3
 0

Ошибки также могут обрабатываться путем повторной попытки не удавшихся вычислений. Ключевые слова аргументы retry_delays и retry_check передаются в retry в качестве ключевых аргументов delays и check соответственно. Если задана пакетная обработка, и весь пакет завершается ошибкой, все элементы пакета повторяются.

Обратите внимание, что если и on_error и retry_delays указаны, хук on_error вызывается до повторной попытки. Если on_error не генерирует (или не перегенерирует) исключение, элемент не будет повторно обрабатываться.

Пример: При ошибках, повторить f для элемента не более 3 раз без задержек между повторениями.

pmap(f, c; retry_delays = zeros(3))

Пример: Повторить f только если исключение не является типом InexactError, с экспоненциально возрастающими задержками до 3 раз. Вернуть NaN вместо всех InexactError случаев.

pmap(f, c; on_error = e->(isa(e, InexactError) ? NaN : rethrow()), retry_delays = ExponentialBackOff(n = 3))

RemoteException(captured)

Исключения при удаленных вычислениях перехватываются и повторно генерируются локально. RemoteException оборачивает pid рабочего процесса и перехваченное исключение. CapturedException перехватывает удаленное исключение и сериализуемую форму стека вызовов, когда было сгенерировано исключение.

ProcessExitedException(worker_id::Int)

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

Future(w::Int, rrid::RRID, v::Union{Some, Nothing}=nothing)

Future — это заполнитель для одного вычисления с неизвестным статусом завершения и временем. Для нескольких потенциальных вычислений см. RemoteChannel. См. remoteref_id для определения AbstractRemoteRef.

RemoteChannel(pid::Integer=myid())

Создайте ссылку на Channel{Any}(1) на процессе pid. По умолчанию pid — текущий процесс.

RemoteChannel(f::Function, pid::Integer=myid())

Создайте ссылки на удаленные каналы определенного размера и типа. f — функция, которая при выполнении на pid должна вернуть реализацию AbstractChannel.

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

По умолчанию pid — текущий процесс.

fetch(x::Future)

Дождитесь и получите значение Future. Полученное значение кэшируется локально. Дальнейшие вызовы fetch к той же ссылке возвращают кэшированное значение. Если удаленное значение — исключение, выбрасывает RemoteException, который перехватывает удаленное исключение и трассировку стека.

fetch(c::RemoteChannel)

Дождитесь и получите значение из RemoteChannel. Исключения обрабатываются так же, как и для Future. Не удаляет полученный элемент.

fetch(x::Any)

Возвратить x.

remotecall(f, id::Integer, args...; kwargs...) -> Future

Асинхронно вызовите функцию f с заданными аргументами на указанном процессе. Возвращает Future. Ключевые аргументы, если таковые имеются, передаются в f.

remotecall_wait(f, id::Integer, args...; kwargs...)

Выполнить более быстрый wait(remotecall(...)) за один сообщение на Worker, указанном идентификатором рабочего процесса id. Ключевые аргументы, если таковые имеются, передаются в f.

См. также wait и remotecall.

remotecall_fetch(f, id::Integer, args...; kwargs...)

Выполнить fetch(remotecall(...)) за одно сообщение. Ключевые аргументы, если таковые имеются, передаются в f. Любые удалённые исключения перехватываются в RemoteException и выбрасываются.

См. также fetch и remotecall.

Примеры

$ julia -p 2

julia> remotecall_fetch(sqrt, 2, 4)
2.0

julia> remotecall_fetch(sqrt, 2, -4)
ERROR: On worker 2:
DomainError with -4.0:
sqrt was called with a negative real argument but will only return a complex result if called with a complex argument. Try sqrt(Complex(x)).
...

remote_do(f, id::Integer, args...; kwargs...) -> nothing

Выполняет f на рабочем процессе id асинхронно. В отличие от remotecall, он не хранит результат вычисления, и нет способа дождаться его завершения.

Успешное выполнение указывает на то, что запрос принят для выполнения на удаленном узле.

Хотя последовательные remotecall к одному и тому же рабочему процессу сериализуются в порядке их вызова, порядок выполнения на удалённом рабочем процессе не определён. Например, remote_do(f1, 2); remotecall(f2, 2); remote_do(f3, 2) сериализует вызов f1, за которым следуют f2 и f3 в этом порядке. Однако не гарантируется, что f1 выполняется до f3 на рабочем процессе 2.

Любые исключения, сгенерированные f, выводятся в stderr на удалённом рабочем процессе.

Ключевые аргументы, если таковые имеются, передаются в f.

put!(rr::RemoteChannel, args...)

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

put!(rr::Future, v)

Сохранить значение в Future rr. Future — ссылки на удаленные значения, записываемые один раз. put! на уже заданном Future выбрасывает Exception. Все асинхронные удалённые вызовы возвращают Future и устанавливают значение возвращаемому значению вызова по завершении.

take!(rr::RemoteChannel, args...)

Получить значение(я) из RemoteChannel rr, удалив значение(я) в процессе.

isready(rr::RemoteChannel, args...)

Определить, есть ли значение в RemoteChannel. Обратите внимание, что эта функция может вызвать гонки, так как к моменту получения результата это может больше не быть истинным. Однако она может безопасно использоваться для Future, так как им присваивается значение только один раз.

isready(rr::Future)

Определить, имеет ли Future сохранённое значение.

Если аргумент Future принадлежит другому узлу, этот вызов будет блокироваться, ожидая ответа. Рекомендуется ожидать rr в отдельной задаче или использовать локальный Channel в качестве прокси:

p = 1
f = Future(p)
errormonitor(@async put!(f, remotecall_fetch(long_computation, p)))
isready(f)  # will not block

AbstractWorkerPool

Супертип для пулов задач, таких как WorkerPool и CachingPool. AbstractWorkerPool должен реализовывать:

• push! - добавить новую задачу в общий пул (доступная + занятая)
• put! - вернуть задачу в доступный пул
• take! - взять задачу из доступного пула (для выполнения удалённой функции)
• length - количество доступных задач в общем пуле
• isready - вернуть false, если добавление take! в пул заблокирует выполнение, иначе true

Стандартные реализации выше (в AbstractWorkerPool) требуют полей

• channel::Channel{Int}
• workers::Set{Int}

где channel содержит идентификаторы свободных задач, а workers — набор всех задач, связанных с этим пулом.

WorkerPool(workers::Union{Vector{Int},AbstractRange{Int}})

Создать WorkerPool из вектора или диапазона идентификаторов задач.

Примеры

$ julia -p 3

julia> WorkerPool([2, 3])
WorkerPool(Channel{Int64}(sz_max:9223372036854775807,sz_curr:2), Set([2, 3]), RemoteChannel{Channel{Any}}(1, 1, 6))

julia> WorkerPool(2:4)
WorkerPool(Channel{Int64}(sz_max:9223372036854775807,sz_curr:2), Set([4, 2, 3]), RemoteChannel{Channel{Any}}(1, 1, 7))

CachingPool(workers::Vector{Int})

Реализация AbstractWorkerPool. remote, remotecall_fetch, pmap (и другие удалённые вызовы, выполняющие функции удалённо) извлекают выгоду из кэширования сериализованных/десериализованных функций на узлах задач, особенно замыканий (которые могут захватывать большие объёмы данных).

Кэш удалённых данных поддерживается на протяжении всего существования возвращаемого CachingPool объекта. Чтобы очистить кэш раньше, используйте clear!(pool).

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

Примеры

const foo = rand(10^8);
wp = CachingPool(workers())
let foo = foo
    pmap(i -> sum(foo) + i, wp, 1:100);
end

Вышеперечисленное передало бы foo только один раз на каждую задачу.

default_worker_pool()

AbstractWorkerPool содержащий свободные workers — используется remote(f) и pmap (по умолчанию). Если не указан явным образом через default_worker_pool!(pool), пул задач по умолчанию инициализируется как WorkerPool.

Примеры

$ julia -p 3

julia> default_worker_pool()
WorkerPool(Channel{Int64}(sz_max:9223372036854775807,sz_curr:3), Set([4, 2, 3]), RemoteChannel{Channel{Any}}(1, 1, 4))

clear!(syms, pids=workers(); mod=Main)

Очищает глобальные привязки в модулях, инициализируя их значения как nothing. syms должен быть типа Symbol или набором Symbol. pids и mod идентифицируют процессы и модули, в которых требуется переинициализация глобальных переменных. Очищаются только те имена, которые определены в mod.

Исключение возникает, если запрашивается очистка глобальной константы.

clear!(pool::CachingPool) -> pool

Удаляет все кэшированные функции со всех участвующих задач.

remote([p::AbstractWorkerPool], f) -> Function

Возвращает анонимную функцию, выполняющую функцию f на доступной задаче (взятой из WorkerPool p если предоставлена) с использованием remotecall_fetch.

remotecall(f, pool::AbstractWorkerPool, args...; kwargs...) -> Future

Вариант remotecall(f, pid, ....), использующий WorkerPool. Ожидает и берёт свободную задачу из pool и выполняет remotecall на ней.

Примеры

$ julia -p 3

julia> wp = WorkerPool([2, 3]);

julia> A = rand(3000);

julia> f = remotecall(maximum, wp, A)
Future(2, 1, 6, nothing)

В этом примере задача выполнялась на задаче с идентификатором 2, вызванной с идентификатором 1.

remotecall_wait(f, pool::AbstractWorkerPool, args...; kwargs...) -> Future

Вариант remotecall_wait(f, pid, ....), использующий WorkerPool. Ожидает и берёт свободную задачу из pool и выполняет remotecall_wait на ней.

Примеры

$ julia -p 3

julia> wp = WorkerPool([2, 3]);

julia> A = rand(3000);

julia> f = remotecall_wait(maximum, wp, A)
Future(3, 1, 9, nothing)

julia> fetch(f)
0.9995177101692958

remotecall_fetch(f, pool::AbstractWorkerPool, args...; kwargs...) -> result

Вариант remotecall_fetch(f, pid, ....), использующий WorkerPool. Ожидает и берёт свободную задачу из pool и выполняет remotecall_fetch на ней.

Примеры

$ julia -p 3

julia> wp = WorkerPool([2, 3]);

julia> A = rand(3000);

julia> remotecall_fetch(maximum, wp, A)
0.9995177101692958

remote_do(f, pool::AbstractWorkerPool, args...; kwargs...) -> nothing

Вариант remote_do(f, pid, ....), использующий WorkerPool. Ожидает и берёт свободную задачу из pool и выполняет remote_do на ней.

@spawn expr

Создаёт замыкание вокруг выражения и запускает его на автоматически выбранном процессе, возвращая Future с результатом. Данный макрос устарел; следует использовать @spawnat :any expr вместо него.

Примеры

julia> addprocs(3);

julia> f = @spawn myid()
Future(2, 1, 5, nothing)

julia> fetch(f)
2

julia> f = @spawn myid()
Future(3, 1, 7, nothing)

julia> fetch(f)
3

@spawnat p expr

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

Примеры

julia> addprocs(3);

julia> f = @spawnat 2 myid()
Future(2, 1, 3, nothing)

julia> fetch(f)
2

julia> f = @spawnat :any myid()
Future(3, 1, 7, nothing)

julia> fetch(f)
3

@fetch expr

Эквивалентно fetch(@spawnat :any expr). См. fetch и @spawnat.

Примеры

julia> addprocs(3);

julia> @fetch myid()
2

julia> @fetch myid()
3

julia> @fetch myid()
4

julia> @fetch myid()
2

@fetchfrom

Эквивалентно fetch(@spawnat p expr). См. fetch и @spawnat.

Примеры

julia> addprocs(3);

julia> @fetchfrom 2 myid()
2

julia> @fetchfrom 4 myid()
4

@distributed

Параллельный цикл для распределённой памяти, имеющий вид:

@distributed [reducer] for var = range
    body
end

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

Обратите внимание, что без функции-редуктора @distributed выполняется асинхронно, т.е. запускает независимые задачи на всех доступных задачах и возвращает результат немедленно, не дожидаясь завершения. Чтобы дождаться завершения, добавьте перед вызовом @sync, например:

@sync @distributed for var = range
    body
end

@everywhere [procs()] expr

Выполняет выражение Main на всех procs. Ошибки в любом из процессов собираются в CompositeException и генерируются. Например:

@everywhere bar = 1

определит Main.bar на всех текущих процессах. Любые добавленные позже процессы (например, с помощью addprocs()) не будут иметь это выражение.

В отличие от @spawnat, @everywhere не захватывает никакие локальные переменные. Вместо этого локальные переменные можно транслировать с помощью интерполяции:

foo = 1
@everywhere bar = $foo

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

Аналогично вызову remotecall_eval(Main, procs, expr), но с двумя дополнительными функциями:

- `using` and `import` statements run on the calling process first, to ensure
  packages are precompiled.
- The current source file path used by `include` is propagated to other processes.

remoteref_id(r::AbstractRemoteRef) -> RRID

Future и RemoteChannel идентифицируются полями:

• where — относится к узлу, где фактически существует подлежащий объект/хранилище, на который ссылается ссылка.

• whence — относится к узлу, из которого была создана удалённая ссылка. Обратите внимание, что это отличается от узла, где фактически существует подлежащий объект ссылки. Например, вызов RemoteChannel(2) из главного процесса приведёт к значению where 2 и значению whence 1.

• id уникален для всех ссылок, созданных из работника, указанного в whence.

В совокупности whence и id однозначно идентифицируют ссылку по всем работникам.

remoteref_id — это API низкого уровня, который возвращает объект RRID, который оборачивает значения whence и id удалённой ссылки.

channel_from_id(id) -> c

API низкого уровня, возвращающий базовый AbstractChannel для id, возвращённого функцией remoteref_id. Вызов допустим только на узле, где существует базовый канал.

worker_id_from_socket(s) -> pid

API низкого уровня, который, получив соединение IO или Worker, возвращает идентификатор pid работника, с которым оно соединено. Это полезно при написании пользовательских методов serialize для типа, оптимизирующих выводимые данные в зависимости от идентификатора процесса получателя.

cluster_cookie() -> cookie

Возвращает cookie кластера.

cluster_cookie(cookie) -> cookie

Устанавливает переданный cookie в качестве cookie кластера, затем возвращает его.

ClusterManager

Супертип для менеджеров кластеров, которые управляют рабочими процессами как кластером. Менеджеры кластеров реализуют, как рабочие процессы могут быть добавлены, удалены и с ними взаимодействовать. SSHManager и LocalManager являются подтипами этого.

WorkerConfig

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

• io — соединение, используемое для доступа к работнику (подтип IO или Nothing)
• host — адрес хоста (либо String, либо Nothing)
• port — порт на хосте, используемый для подключения к работнику (либо Int, либо Nothing)

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

• count — количество рабочих процессов, подлежащих запуску на хосте
• exename — путь к исполняемому файлу Julia на хосте, по умолчанию "$(Sys.BINDIR)/julia" или "$(Sys.BINDIR)/julia-debug"
• exeflags — флаги, используемые при удалённом запуске Julia

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

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

• tunnel — true (использовать туннелирование), false (не использовать туннелирование), или nothing (использовать значение по умолчанию для менеджера)
• multiplex — true (использовать многократное использование SSH для туннелирования) или false
• forward — опция переадресации, используемая для опции -L ssh
• bind_addr — адрес на удалённом хосте для привязки
• sshflags — флаги, используемые при установлении соединения SSH
• max_parallel — максимальное количество рабочих процессов для одновременного подключения к хосту

Некоторые поля используются как менеджерами LocalManager, так и SSHManager:

• connect_at — определяет, является ли это вызов «работник-работник» или «драйвер-работник»
• process — процесс, который будет подключён (обычно менеджер присваивает его во время вызова addprocs)
• ospid — идентификатор процесса согласно операционной системе хоста, используется для прерывания рабочих процессов
• environ — частный словарь, используемый для хранения временной информации менеджерами Local/SSH
• ident — работник, идентифицированный менеджером ClusterManager
• connect_idents — список идентификаторов рабочих процессов, к которым работник должен подключиться при использовании пользовательской топологии
• enable_threaded_blas — true, false, или nothing, использовать ли потоковую BLAS в рабочих процессах

launch(manager::ClusterManager, params::Dict, launched::Array, launch_ntfy::Condition)

Реализуется менеджерами кластеров. Для каждого процесса Julia, запущенного этой функцией, она должна добавить запись WorkerConfig в launched и уведомить launch_ntfy. Функция ДОЛЖНА завершиться после запуска всех рабочих процессов, запрошенных в manager. params — словарь всех ключевых аргументов, с которыми был вызван addprocs.

manage(manager::ClusterManager, id::Integer, config::WorkerConfig. op::Symbol)

Реализуется менеджерами кластеров. Она вызывается в главном процессе во время существования рабочего процесса с соответствующими значениями op:

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

kill(manager::ClusterManager, pid::Int, config::WorkerConfig)

Реализуется менеджерами кластеров. Вызывается в главном процессе функцией rmprocs. Она должна вызвать выход удалённого рабочего процесса, указанного в pid. kill(manager::ClusterManager.....) выполняет удалённый exit() в pid.

connect(manager::ClusterManager, pid::Int, config::WorkerConfig) -> (instrm::IO, outstrm::IO)

Реализуется менеджерами кластеров с использованием пользовательского транспорта. Она должна установить логическое соединение с работником с идентификатором pid, указанным в config, и вернуть пару объектов IO. Сообщения от pid к текущему процессу будут считываться из instrm, в то время как сообщения, отправляемые в pid, будут записываться в outstrm. Реализация пользовательского транспорта должна гарантировать, что сообщения передаются и принимаются полностью и в порядке. connect(manager::ClusterManager.....) устанавливает соединения TCP/IP между рабочими процессами.

init_worker(cookie::AbstractString, manager::ClusterManager=DefaultClusterManager())

Вызывается менеджерами кластеров, реализующими пользовательский транспорт. Она инициализирует только что запущенный процесс как работник. Аргумент командной строки --worker[=<cookie>] имеет эффект инициализации процесса как работника с использованием сокетов TCP/IP для транспорта. cookie является cluster_cookie.

start_worker([out::IO=stdout], cookie::AbstractString=readline(stdin); close_stdin::Bool=true, stderr_to_stdout::Bool=true)

start_worker — это внутренняя функция, являющаяся точкой входа по умолчанию для рабочих процессов, подключающихся через TCP/IP. Она настраивает процесс как работника кластера Julia.

Информация о хосте:порту записывается в поток out (по умолчанию — stdout).

Функция считывает cookie из stdin, если необходимо, и прослушивает свободный порт (или, если указано, порт в параметре командной строки --bind-to). Она планирует задачи для обработки входящих подключений и запросов TCP. Она также (при необходимости) закрывает stdin и перенаправляет stderr в stdout.

Функция не возвращает значение.

process_messages(r_stream::IO, w_stream::IO, incoming::Bool=true)

Вызывается менеджерами кластеров, использующими пользовательский транспорт. Она должна вызываться, когда пользовательская реализация транспорта получает первое сообщение от удалённого работника. Пользовательский транспорт должен управлять логическим соединением с удалённым работником и предоставлять два объекта IO, один для входящих сообщений и другой для сообщений, предназначенных для удалённого работника. Если incoming равно true, удалённый узел инициировал подключение. Тот, кто из пары инициировал подключение, отправляет cookie кластера и номер версии Julia для выполнения рукопожатия аутентификации.

См. также cluster_cookie.

default_addprocs_params(mgr::ClusterManager) -> Dict{Symbol, Any}

Реализуется менеджерами кластеров. По умолчанию параметры ключевых слов, передаваемые при вызове addprocs(mgr). Минимальный набор опций доступен при вызове default_addprocs_params()