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

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

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

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

Параметры

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

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

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

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

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

Параметры

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

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

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

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

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

Обратите внимание, что эта функция требует, чтобы у диспетчера задач был параметр :temporary в качестве параметра :restart (по умолчанию), поскольку async_nolink/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/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, _from, %{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_call(:start_task, _from, %{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

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

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

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

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

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

@spec async_stream(Supervisor.supervisor(), Enumerable.t(), (term() -> term()), [
  async_stream_option()
]) ::
  Enumerable.t()

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

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

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

@spec async_stream(
  Supervisor.supervisor(),
  Enumerable.t(),
  module(),
  atom(),
  [term()],
  [
    async_stream_option()
  ]
) :: Enumerable.t()

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

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

При потоковой передаче каждая задача будет излучать {: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}.

• :zip_input_on_exit - (с версии v1.14.0) добавляет исходный вход в кортежи :exit. Выдаваемое значение для этой задачи — {:exit, {input, reason}}, где input — элемент коллекции, вызвавший выход из строя во время обработки. По умолчанию false.

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

Примеры

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

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

@spec async_stream_nolink(
  Supervisor.supervisor(),
  Enumerable.t(),
  (term() -> term()),
  [
    async_stream_option()
  ]
) :: Enumerable.t()

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

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

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

Обработка ошибок и очистка

Даже если задачи не связаны с вызывающим процессом, нет риска оставить зависшие задачи, работающие после остановки потока.

Рассмотрим следующий пример:

Task.Supervisor.async_stream_nolink(MySupervisor, collection, fun, on_timeout: :kill_task, ordered: false)
|> Enum.each(fn
  {:ok, _} -> :ok
  {:exit, reason} -> raise "Task exited: #{Exception.format_exit(reason)}"
end)

Если одна задача вызывает исключение или истекает время ожидания:

1. выполняется второй блок кода
2. вызывается исключение
3. поток останавливается
4. все активные задачи будут остановлены

Вот ещё один пример:

Task.Supervisor.async_stream_nolink(MySupervisor, collection, fun, on_timeout: :kill_task, ordered: false)
|> Stream.filter(&match?({:ok, _}, &1))
|> Enum.take(3)

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

Просто выполнение потока с помощью Stream.run/1 с другой стороны проигнорирует ошибки и обработает весь поток.

@spec async_stream_nolink(
  Supervisor.supervisor(),
  Enumerable.t(),
  module(),
  atom(),
  [term()],
  [
    async_stream_option()
  ]
) :: Enumerable.t()

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

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

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

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

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

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

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

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

Task.Supervisor.start_child(MyTaskSupervisor, fn ->
  IO.puts "I am running in a task"
end)

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

Параметры

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

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

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

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

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

@spec 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.

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

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