Task.Supervisor

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

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

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

children = [
  {Task.Supervisor, name: MyApp.TaskSupervisor}
]

Supervisor.start_link(strategy: :one_for_one)

Опции, заданные в спецификации дочерней задачи, документированы в start_link/1.

См. модуль Task для получения дополнительных примеров.

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

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

Обзор

Типы

option(): Значения опций, используемые start_link

Функции

async(supervisor, fun, options \\ []): Запускает задачу, на которой можно ожидать результата
async(supervisor, module, fun, args, options \\ []): Запускает задачу, на которой можно ожидать результата
async_nolink(supervisor, fun, options \\ []): Запускает задачу, на которой можно ожидать результата
async_nolink(supervisor, module, fun, args, options \\ []): Запускает задачу, на которой можно ожидать результата
async_stream(supervisor, enumerable, fun, options \\ []): Возвращает поток, который выполняет заданную функцию fun конкурирующим образом для каждого элемента в 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): Возвращает все PID дочерних задач
start_child(supervisor, fun, options \\ []): Запускает задачу как дочернюю задачу заданного supervisor
start_child(supervisor, module, fun, args, options \\ []): Запускает задачу как дочернюю задачу заданного supervisor
start_link(options \\ []): Запускает нового наблюдателя
terminate_child(supervisor, pid): Прерывает дочернюю задачу с указанным pid

Типы

option()

option() ::
  Supervisor.option()
  | {:restart, :supervisor.restart()}
  | {:shutdown, :supervisor.shutdown()}

Значения опций, используемые start_link

Функции

async(supervisor, fun, options \\ [])

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

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

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

Опции

:shutdown - :brutal_kill если задачи должны быть убиты непосредственно при завершении работы, или целое число, указывающее значение таймаута, по умолчанию 5000 миллисекунд.

async(supervisor, module, fun, args, options \\ [])

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

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

Опции

:shutdown - :brutal_kill если задачи должны быть убиты непосредственно при завершении работы, или целое число, указывающее значение таймаута, по умолчанию 5000 миллисекунд.

async_nolink(supervisor, fun, options \\ [])

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

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

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

Опции

:shutdown - :brutal_kill если задачи должны быть убиты непосредственно при завершении работы, или целое число, указывающее значение таймаута, по умолчанию 5000 миллисекунд.

Совместимость с поведением 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, options \\ [])

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

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

Обратите внимание, что этой функции требуется, чтобы наблюдатель задач имел :temporary в качестве опции :restart (по умолчанию), так как async_nolink/4 сохраняет прямую ссылку на задачу, которая теряется, если задача перезапускается.

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

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

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

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

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

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

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

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

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

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

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

И, наконец, если вы обнаружите, что ловите выходные данные для обработки исключений внутри потокового преобразования, рассмотрите использование async_stream_nolink/6 для запуска задач, которые не связаны с текущим процессом.

Опции

:max_concurrency - устанавливает максимальное количество задач, которые будут выполняться одновременно. По умолчанию равно System.schedulers_online/0.
:ordered - указывает, должны ли результаты возвращаться в том же порядке, что и входной поток. Эта опция полезна при работе с большими потоками и позволяет избежать буферизации результатов до их доставки. По умолчанию true.
:timeout - максимальное время ожидания (в миллисекундах) без получения ответа от задачи (по всем выполняющимся задачам). По умолчанию 5000.
:on_timeout - что делать, когда задача превышает время ожидания. Возможные значения:
- :exit (по умолчанию) - процесс, создавший задачи, завершается.
- :kill_task - задача, превысившая время ожидания, убивается. Выводимое значение для этой задачи - {:exit, :timeout}.
:shutdown - :brutal_kill если задачи должны быть убиты непосредственно при завершении работы, или целое число, указывающее значение таймаута, по умолчанию 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()
) :: Enumerable.t()

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

Каждый элемент в enumerable передаётся в качестве аргумента указанной функции fun и обрабатывается своей задачей. Задачи будут созданы под заданным 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()
) :: Enumerable.t()

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

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

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

children(supervisor)

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

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

start_child(supervisor, fun, options \\ [])

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

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

Опции

:restart — стратегия перезапуска, может быть :temporary (по умолчанию), :transient или :permanent. :temporary означает, что задача никогда не перезапускается, :transient означает, что она перезапускается, если выход не :normal, :shutdown или {:shutdown, reason}. Стратегия перезапуска :permanent означает, что она всегда перезапускается. По умолчанию :temporary.
:shutdown — :brutal_kill если задачи должны быть убиты при завершении работы, или целое число, указывающее значение таймаута, по умолчанию 5000 миллисекунд.

start_child(supervisor, module, fun, args, options \\ [])

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

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

start_link(options \\ [])

start_link([option()]) :: Supervisor.on_start()

Запускает новый супервайзер.

Примеры

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

{Task.Supervisor, name: MyApp.TaskSupervisor}

Вы также можете запустить его, вызвав start_link/1 напрямую:

Task.Supervisor.start_link(name: MyApp.TaskSupervisor)

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

Опции

:name — используется для регистрации имени супервайзера, поддерживаемые значения описаны в разделе Name Registration в документации модуля GenServer;
:max_restarts, :max_seconds и :max_children — как указано в DynamicSupervisor;

Эта функция также может принимать :restart и :shutdown в качестве опций, но эти две опции устарели, и теперь предпочтительнее передавать их непосредственно в start_child и async.

terminate_child(supervisor, pid)

terminate_child(Supervisor.supervisor(), pid()) :: :ok | {:error, :not_found}

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

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