Task.Supervisor

Наблюдатель за задачами.

Этот модуль определяет наблюдателя, который может использоваться для динамического наблюдения за задачами. Под капотом этот модуль реализован как :simple_one_for_one наблюдатель, где рабочие по умолчанию являются временными (то есть, они не перезапускаются после завершения; ознакомьтесь с документацией по start_link/1 для получения дополнительной информации о выборе стратегии перезапуска).

Для получения дополнительной информации см. модуль Task.

Регистрация имени

Task.Supervisor привязан к тем же правилам регистрации имен, что и GenServer. Дополнительную информацию см. в документации по GenServer.

Краткое описание

Функции

async(supervisor, fun): Запускает задачу, на которой можно ожидать результата
async(supervisor, module, fun, args): Запускает задачу, на которой можно ожидать результата
async_nolink(supervisor, fun): Запускает задачу, на которой можно ожидать результата
async_nolink(supervisor, module, fun, args): Запускает задачу, на которой можно ожидать результата
async_stream(supervisor, enumerable, fun, options \\ []): Возвращает поток, который выполняет указанную function одновременно для каждого элемента в enumerable
async_stream(supervisor, enumerable, module, function, args, options \\ []): Возвращает поток, который выполняет указанную module, function и args одновременно для каждого элемента в enumerable
async_stream_nolink(supervisor, enumerable, fun, options \\ []): Возвращает поток, который выполняет указанную function одновременно для каждого элемента в enumerable
async_stream_nolink(supervisor, enumerable, module, function, args, options \\ []): Возвращает поток, который выполняет указанную module, function и args одновременно для каждого элемента в enumerable
children(supervisor): Возвращает все идентификаторы процессов дочерних задач
start_child(supervisor, fun): Запускает задачу как дочернюю задачу для данного supervisor
start_child(supervisor, module, fun, args): Запускает задачу как дочернюю задачу для данного supervisor
start_link(opts \\ []): Запускает нового наблюдателя
terminate_child(supervisor, pid): Завершает дочернюю задачу с заданным pid

Функции

async(supervisor, fun)

async(Supervisor.supervisor(), (... -> any())) :: Task.t()

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

supervisor должно быть ссылкой, как определено в Task.Supervisor. Задача всё ещё будет связана с вызывающим её процессом, см. Task.async/3 для дополнительной информации и async_nolink/2 для варианта без связи.

async(supervisor, module, fun, args)

async(Supervisor.supervisor(), module(), atom(), [term()]) :: Task.t()

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

async_nolink(supervisor, fun)

async_nolink(Supervisor.supervisor(), (... -> any())) :: Task.t()

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

supervisor должно быть ссылкой, как определено в Task.Supervisor. Задача не будет связана с вызывающим её процессом, см. Task.async/3 для дополнительной информации.

Совместимость с поведением OTP

Если вы создаёте задачу с использованием async_nolink внутри поведения OTP, например, GenServer, вы должны обрабатывать сообщение, пришедшее от задачи, внутри вашей функции обратного вызова GenServer.handle_info/2.

Ответ, отправленный задачей, будет иметь формат {ref, result}, где ref — ссылка на монитор, хранящаяся в структуре задачи, а result — возвращаемое значение функции задачи.

Обратите внимание, что независимо от того, как завершается задача, созданная с использованием async_nolink, вызывающий процесс всегда получит сообщение :DOWN с тем же значением ref, которое хранится в структуре задачи. Если задача завершилась успешно, причина в сообщении :DOWN будет :normal.

async_nolink(supervisor, module, fun, args)

async_nolink(Supervisor.supervisor(), module(), atom(), [term()]) :: Task.t()

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

async_stream(supervisor, enumerable, fun, options \\ [])

async_stream(Supervisor.supervisor(), Enumerable.t(), (term() -> term()), Keyword.t()) :: Enumerable.t()

Возвращает поток, который выполняет указанную function одновременно для каждого элемента в enumerable.

Каждый элемент будет добавлен в заданный args и обработан собственной задачей. Задачи будут запущены под заданным supervisor и связаны с текущим процессом, аналогично async/2.

См. async_stream/6 для обсуждения и примеров.

async_stream(supervisor, enumerable, module, function, args, options \\ [])

async_stream(Supervisor.supervisor(), Enumerable.t(), module(), atom(), [term()], Keyword.t()) :: Enumerable.t()

Возвращает поток, который выполняет указанную module, function и args одновременно для каждого элемента в enumerable.

При потоковой передаче каждая задача будет генерировать {:ok, val} при успешном завершении или {:exit, val} , если вызывающий процесс обрабатывает завершения. Результаты генерируются в том же порядке, что и исходный enumerable.

Уровень параллелизма можно контролировать с помощью опции :max_concurrency и он по умолчанию равен System.schedulers_online/0. Таймаут также можно указать в качестве опции, и по умолчанию он равен 5000, и он по умолчанию равен максимальному времени ожидания ответа от задачи.

Наконец, если вы обрабатываете завершения внутри потока async, рассмотрите использование async_stream_nolink/6 для запуска задач, не связанных с текущим процессом.

Опции

:max_concurrency - устанавливает максимальное количество задач, выполняемых одновременно. По умолчанию равен System.schedulers_online/0.
:timeout - максимальное время ожидания ответа от задачи (для всех запущенных задач). По умолчанию равен 5000.

Примеры

Давайте создадим поток, а затем перечислим его:

stream = Task.Supervisor.async_stream(MySupervisor, collection, Mod, :expensive_fun, [])
Enum.to_list(stream)

async_stream_nolink(supervisor, enumerable, fun, options \\ [])

async_stream_nolink(Supervisor.supervisor(), Enumerable.t(), (term() -> term()), Keyword.t()) :: Enumerable.t()

Возвращает поток, который выполняет указанную function одновременно для каждого элемента в enumerable.

Каждый элемент будет добавлен в заданный args и обработан собственной задачей. Задачи будут запущены под заданным supervisor и не связаны с текущим процессом, аналогично async_nolink/2.

См. async_stream/6 для обсуждения и примеров.

async_stream_nolink(supervisor, enumerable, module, function, args, options \\ [])

async_stream_nolink(Supervisor.supervisor(), Enumerable.t(), module(), atom(), [term()], Keyword.t()) :: Enumerable.t()

Возвращает поток, который выполняет указанную module, function и args одновременно для каждого элемента в enumerable.

См. async_stream/6 для обсуждения и примеров.

children(supervisor)

children(Supervisor.supervisor()) :: [pid()]

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

start_child(supervisor, fun)

start_child(Supervisor.supervisor(), (... -> any())) :: {:ok, pid()}

Запускает задачу как дочернюю задачу для данного supervisor.

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

start_child(supervisor, module, fun, args)

start_child(Supervisor.supervisor(), module(), atom(), [term()]) :: {:ok, pid()}

Запускает задачу как дочернюю задачу для данного supervisor.

Аналогично start_child/2, за исключением того, что задача задаётся с помощью указанных module, fun и args.

start_link(opts \\ [])

start_link(Supervisor.options()) :: Supervisor.on_start()

Запускает нового супервизора.

Поддерживаемые опции:

:name — используется для регистрации имени супервизора, допустимые значения описаны в разделе Name Registration в документации модуля GenServer;
:restart — стратегия перезапуска, может быть :temporary (по умолчанию), :transient или :permanent. Подробнее см. в Supervisor.Spec. По умолчанию :temporary, поэтому задачи не перезапускаются автоматически при завершении работы или при сбоях;
:shutdown — :brutal_kill, если задачи должны быть убиты непосредственно при выключении, или целое число, обозначающее время ожидания, по умолчанию 5000 миллисекунд;
:max_restarts и :max_seconds — как указано в Supervisor.Spec.supervise/2;

terminate_child(supervisor, pid)

terminate_child(Supervisor.supervisor(), pid()) :: :ok

Прерывает дочерний процесс с заданным pid.

© 2012 Plataformatec
Licensed under the Apache License, Version 2.0.
https://hexdocs.pm/elixir/1.4.5/Task.Supervisor.html