Spec-Zone.ru › Elixir 1.8

Task.Supervisor

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

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

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

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

Supervisor.start_link(children, strategy: :one_for_one)

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

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

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

Наблюдатель связан с теми же правилами регистрации имен, что и 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) отображается одновременно на каждом элементе в enumerable.

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

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

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

Возвращает поток, где заданная функция (module и function) отображается одновременно на каждом элементе в enumerable.

children(supervisor)

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

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

Возникает ошибка, если supervisor достиг максимального количества дочерних задач.

Опции

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

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

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

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

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

Возникает ошибка, если supervisor достиг максимального количества дочерних задач.

Опции

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

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

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

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

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

Возникает ошибка, если supervisor достиг максимального количества дочерних задач.

Опции

  • :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/3, когда есть обоснованное ожидание того, что задача может потерпеть неудачу, и вы не хотите, чтобы она приводила к сбою вызывающего её.

Давайте посмотрим пример, где GenServer предназначен для запуска одной задачи и отслеживания её состояния:

defmodule MyApp.Server do
  use GenServer

  # ...

  def start_task do
    GenServer.call(__MODULE__, :start_task)
  end

  # In this case the task is already running, so we just return :ok.
  def handle_call(:start_task, %{ref: ref} = state) when is_reference(ref) do
    {:reply, :ok, state}
  end

  # The task is not running yet, so let's start it.
  def handle_cast(:start_task, %{ref: nil} = state) do
    task =
      Task.Supervisor.async_nolink(MyApp.TaskSupervisor, fn ->
        ...
      end)

    # We return :ok and the server will continue running
    {:reply, :ok, %{state | ref: task.ref}}
  end

  # The task completed successfully
  def handle_info({ref, answer}, %{ref: ref} = state) do
    # We don't care about the DOWN message now, so let's demonitor and flush it
    Process.demonitor(ref, [:flush])
    # Do something with the result and then return
    {:noreply, %{state | ref: nil}}
  end

  # The task failed
  def handle_info({:DOWN, ref, :process, _pid, _reason}, %{ref: ref} = state) do
    # Log and possibly restart the task...
    {:noreply, %{state | ref: nil}}
  end
end

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

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

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

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

Возникает ошибка, если supervisor достиг максимального количества дочерних задач.

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

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

(since 1.4.0)
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 \\ [])

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

Возвращает поток, где заданная функция (module и function) отображается одновременно на каждом элементе в enumerable.

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

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

Уровень параллельности и время, которое задачи могут работать, можно контролировать с помощью опций (см. раздел "Опции" ниже).

Если вы обнаружите, что ловите завершения для обработки завершений внутри потоковой передачи, рассмотрите возможность использования 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 \\ [])

(с 1.4.0)
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 \\ [])

(с 1.4.0)
async_stream_nolink(
  Supervisor.supervisor(),
  Enumerable.t(),
  module(),
  atom(),
  [term()],
  keyword()
) :: Enumerable.t()

Возвращает поток, где заданная функция (module и function) отображается асинхронно на каждом элементе в enumerable.

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

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

children(supervisor) 

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

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

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

start_child(Supervisor.supervisor(), (() -> any()), keyword()) ::
  DynamicSupervisor.on_start_child()

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

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

Параметры

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

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

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

start_child(Supervisor.supervisor(), module(), atom(), [term()], keyword()) ::
  DynamicSupervisor.on_start_child()

Запускает задачу как дочернюю задачу данного 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.8.2/Task.Supervisor.html

Spec-Zone.ru

Настройки Оффлайн Что нового Помощь О нас
Spec-Zone .ru
спецификации, руководства, описания, API