Задача

Структура Task.

Она содержит следующие поля:

• :mfa - кортеж из трёх элементов, содержащий модуль, имя функции и арность, вызываемые для запуска задачи в async/1 и async/3

• :owner - PID процесса, который запустил задачу

• :pid - PID процесса задачи; nil если для задачи нет специально назначенного процесса

• :ref - непрозрачный термин, используемый в качестве ссылки монитора задачи

@spec async((-> any())) :: t()

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

fun должна быть анонимной функцией с нулевой арностью. Эта функция порождает процесс, который связан и отслеживается процессом-вызывающим. Возвращается структура Task с соответствующей информацией.

Если вы запускаете async, вы обязаны дождаться её завершения. Это делается вызовом Task.await/2 или Task.yield/2, за которым следует Task.shutdown/2 для возвращённой задачи. В качестве альтернативы, если вы запускаете задачу внутри GenServer, то GenServer автоматически будет ждать завершения и вызывать GenServer.handle_info/2 с ответом задачи и связанным :DOWN сообщением.

См. документацию модуля Task для получения дополнительной информации об общем использовании асинхронных задач.

Связывание

Эта функция порождает процесс, который связан и отслеживается вызывающим процессом. Важно, что эта связь прерывает задачу, если родительский процесс завершается. Также она гарантирует, что код до async/await будет иметь те же свойства и после добавления асинхронного вызова. Например, представьте:

x = heavy_fun()
y = some_fun()
x + y

Теперь вы хотите сделать heavy_fun() асинхронной:

x = Task.async(&heavy_fun/0)
y = some_fun()
Task.await(x) + y

Как и прежде, если heavy_fun/0 завершится ошибкой, вся вычислительная задача завершится ошибкой, включая вызывающий процесс. Если вы не хотите, чтобы задача завершалась ошибкой, нужно изменить код heavy_fun/0 аналогично тому, как вы бы это сделали без асинхронного вызова. Например, возвращая {:ok, val} | :error результаты или, в более сложных случаях, используя try/rescue. Другими словами, асинхронная задача должна рассматриваться как расширение вызывающего процесса, а не как механизм для изоляции его от всех ошибок.

Если вы не хотите связывать вызывающий процесс с задачей, используйте управляемую задачу с Task.Supervisor и вызовите Task.Supervisor.async_nolink/2.

В любом случае, избегайте следующего:

• Установка :trap_exit в true - перехват выхода должен использоваться только в особых случаях, так как это сделает ваш процесс неуязвимым не только к выходам из задачи, но и от других процессов.

  Более того, даже при перехвате выходов, вызов await всё ещё завершится ошибкой, если задача завершилась без возвращения результата.

• Отключение связи процесса задачи, запущенного с async/await. Если вы отключите связь процессов, и задача не принадлежит ни одному диспетчеру, могут остаться зависшие задачи в случае завершения вызывающего процесса.

Метаданные

Задача, созданная этой функцией, хранит :erlang.apply/2 в поле метаданных :mfa, которое используется внутри для применения анонимной функции. Используйте async/3, если вы хотите использовать другую функцию в качестве метаданных.

@spec async(module(), atom(), [term()]) :: t()

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

Аналогично async/1, за исключением того, что функция, которую необходимо запустить, задаётся с помощью указанных module, function_name, и args. module, function_name, и его арность хранятся в поле :mfa для целей отражения.

@spec async_stream(Enumerable.t(), (term() -> term()), keyword()) :: Enumerable.t()

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

Работает так же, как async_stream/5, но с анонимной функцией вместо кортежа модуль-функция-аргументы. fun должна быть анонимной функцией с одной арностью.

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

Пример

Вычислить количество кодовых точек в каждой строке асинхронно, затем сложить счётчики с помощью reduce.

iex> strings = ["long string", "longer string", "there are many of these"]
iex> stream = Task.async_stream(strings, fn text -> text |> String.codepoints() |> Enum.count() end)
iex> Enum.reduce(stream, 0, fn {:ok, num}, acc -> num + acc end)
47

См. async_stream/5 для обсуждения, опций и дополнительных примеров.

@spec async_stream(Enumerable.t(), module(), atom(), [term()], keyword()) ::
  Enumerable.t()

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

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

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

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

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

Опции

• :max_concurrency — устанавливает максимальное количество задач, выполняемых одновременно. По умолчанию System.schedulers_online/0.

• :ordered — определяет, должны ли результаты возвращаться в том же порядке, что и входной поток. При упорядоченном выводе Elixir может потребоваться буферизовать результаты, чтобы выводить их в исходном порядке. Установка этого параметра в false отключает буферизацию, но также удаляет упорядоченность. Это также полезно, если вам нужны только побочные эффекты задач. Обратите внимание, что независимо от значения :ordered, задачи будут обрабатываться асинхронно. Если вам нужно обрабатывать элементы в порядке, рассмотрите использование Enum.map/2 или Enum.each/2 вместо этого. По умолчанию true.

• :timeout — максимальное время (в миллисекундах или :infinity) выполнения каждой задачи. По умолчанию 5000.

• :on_timeout — действия при истечении времени ожидания задачи. Возможные значения:

  • :exit (по умолчанию) — вызывающий процесс завершается.
  • :kill_task — задача, превысившая лимит времени, убивается. Выдаваемое значение для этой задачи {:exit, :timeout}.

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

Пример

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

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

Конкурентность можно увеличить или уменьшить, используя опцию :max_concurrency. Например, если задачи связаны с вводом-выводом, значение можно увеличить:

max_concurrency = System.schedulers_online() * 2
stream = Task.async_stream(collection, Mod, :expensive_fun, [], max_concurrency: max_concurrency)
Enum.to_list(stream)

Если результат вычисления не важен, вы можете запустить поток с помощью Stream.run/1. Также установите ordered: false, так как порядок результатов тоже не важен:

stream = Task.async_stream(collection, Mod, :expensive_fun, [], ordered: false)
Stream.run(stream)

Первые асинхронные задачи, которые завершаются

Вы также можете использовать async_stream/3 для выполнения M задач и поиска N завершенных задач. Например:

[
  &heavy_call_1/0,
  &heavy_call_2/0,
  &heavy_call_3/0
]
|> Task.async_stream(fn fun -> fun.() end, ordered: false, max_concurrency: 3)
|> Stream.filter(&match?({:ok, _}, &1))
|> Enum.take(2)

В приведенном примере мы выполняем три задачи и ждем завершения первых двух. Мы используем Stream.filter/2 для ограничения себя только успешно завершенными задачами и Enum.take/2 для получения N элементов. Важно установить как ordered: false, так и max_concurrency: M, где M — количество задач, чтобы убедиться, что все вызовы выполняются параллельно.

Внимание: неограниченное async + take

Если вы хотите потенциально обработать большое количество элементов и сохранить только часть результатов, вы можете обработать больше элементов, чем нужно. Давайте рассмотрим пример:

1..100
|> Task.async_stream(fn i ->
  Process.sleep(100)
  IO.puts(to_string(i))
end)
|> Enum.take(10)

Запуск примера на машине с 8 ядрами обработает 16 элементов, даже если вам нужны только 10, так как async_stream/3 обрабатывает элементы параллельно. Это происходит потому, что он обрабатывает 8 элементов одновременно. Затем все 8 элементов завершаются примерно в одно и то же время, что приводит к запуску обработки еще 8 элементов. Из этих дополнительных 8 элементов будут использованы только 2, а остальные будут завершены.

В зависимости от задачи вы можете отфильтровать или ограничить количество элементов заранее:

1..100
|> Stream.take(10)
|> Task.async_stream(fn i ->
  Process.sleep(100)
  IO.puts(to_string(i))
end)
|> Enum.to_list()

В других случаях, вероятно, нужно будет настроить :max_concurrency для ограничения количества элементов, которые могут быть обработаны сверх меры, за счет уменьшения конкурентности. Вы также можете установить количество элементов для взятия, кратное :max_concurrency. Например, установив max_concurrency: 5 в приведенном выше примере.

@spec await(t(), timeout()) :: term()

Ожидает ответа задачи и возвращает его.

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

Время ожидания (в миллисекундах или :infinity) может быть задано со значением по умолчанию 5000. Если время ожидания истекло, вызывающий процесс завершится. Если процесс задачи связан с вызывающим процессом, что происходит, когда задача запущена с помощью async, то процесс задачи также завершится. Если процесс задачи ловит исключения или не связан с вызывающим процессом, он продолжит работу.

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

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

Примеры

iex> task = Task.async(fn -> 1 + 1 end)
iex> Task.await(task)
2

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

Не рекомендуется await длительно выполняющуюся задачу внутри поведения OTP, например, GenServer. Вместо этого вы должны сопоставить сообщение, поступающее от задачи, в вашем обратном вызове GenServer.handle_info/2.

GenServer получит два сообщения по handle_info/2:

• {ref, result} — сообщение ответа, где ref ссылка на монитор, возвращенная task.ref, а result результат задачи

• {:DOWN, ref, :process, pid, reason} — поскольку все задачи также отслеживаются, вы также получите сообщение :DOWN, отправленное Process.monitor/1. Если вы получите сообщение :DOWN без ответа, это означает, что задача завершилась ошибкой

Также следует учитывать, что задачи, запущенные с помощью Task.async/1, всегда связаны с вызывающими процессами, и вам, возможно, не захочется, чтобы GenServer завершился, если задача завершится ошибкой. Поэтому предпочтительнее использовать Task.Supervisor.async_nolink/3 внутри поведения OTP. Для полноты, вот пример GenServer, который запускает задачи и обрабатывает их результаты:

defmodule GenServerTaskExample do
  use GenServer

  def start_link(opts) do
    GenServer.start_link(__MODULE__, :ok, opts)
  end

  def init(_opts) do
    # We will keep all running tasks in a map
    {:ok, %{tasks: %{}}}
  end

  # Imagine we invoke a task from the GenServer to access a URL...
  def handle_call(:some_message, _from, state) do
    url = ...
    task = Task.Supervisor.async_nolink(MyApp.TaskSupervisor, fn -> fetch_url(url) end)

    # After we start the task, we store its reference and the url it is fetching
    state = put_in(state.tasks[task.ref], url)

    {:reply, :ok, state}
  end

  # If the task succeeds...
  def handle_info({ref, result}, state) do
    # The task succeed so we can demonitor its reference
    Process.demonitor(ref, [:flush])

    {url, state} = pop_in(state.tasks[ref])
    IO.puts "Got #{inspect(result)} for URL #{inspect url}"
    {:noreply, state}
  end

  # If the task fails...
  def handle_info({:DOWN, ref, _, _, reason}, state) do
    {url, state} = pop_in(state.tasks[ref])
    IO.puts "URL #{inspect url} failed with reason #{inspect(reason)}"
    {:noreply, state}
  end
end

После определения сервера вы захотите запустить диспетчера задач выше и GenServer в вашей дереве надзора:

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

Supervisor.start_link(children, strategy: :one_for_one)

@spec await_many([t()], timeout()) :: [term()]

Ожидает ответы от нескольких задач и возвращает их.

Эта функция получает список задач и ожидает их ответы в заданном временном интервале. Она возвращает список результатов в том же порядке, что и задачи, предоставленные в tasks аргументе.

Если какой-либо из процессов задач завершается аварийно, вызывающий процесс завершится по той же причине, что и эта задача.

Таймаут в миллисекундах или :infinity, может быть задан со значением по умолчанию 5000. Если таймаут истечёт, вызывающий процесс завершится. Любые процессы задач, связанные с вызывающим процессом (что имеет место, когда задача запускается с async) также завершатся. Любые процессы задач, которые обрабатывают завершения или не связаны с вызывающим процессом, продолжат выполнение.

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

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

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

Не рекомендуется await длительные задачи внутри поведения OTP, такого как GenServer. См. await/2 для получения дополнительной информации.

Примеры

iex> tasks = [
...>   Task.async(fn -> 1 + 1 end),
...>   Task.async(fn -> 2 + 3 end)
...> ]
iex> Task.await_many(tasks)
[2, 5]

@spec child_spec(term()) :: Supervisor.child_spec()

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

arg передаётся как аргумент в Task.start_link/1 в :start поле спецификации.

Для получения дополнительной информации, см. модуль Supervisor, функцию Supervisor.child_spec/2 и тип Supervisor.child_spec/0.

@spec completed(any()) :: t()

Запускает задачу, которая сразу же завершается с указанным result.

В отличие от async/1, эта задача не порождает связанный процесс. Её можно ожидать или вызывать, как и любую другую задачу.

Использование

В некоторых случаях полезно создать задачу «completed», которая представляет задачу, которая уже выполнилась и сгенерировала результат. Например, при обработке данных вы можете определить, что некоторые входные данные неверны, прежде чем передавать их для дальнейшей обработки:

def process(data) do
  tasks =
    for entry <- data do
      if invalid_input?(entry) do
        Task.completed({:error, :invalid_input})
      else
        Task.async(fn -> further_process(entry) end)
      end
    end

  Task.await_many(tasks)
end

Во многих случаях Task.completed/1 можно избежать в пользу непосредственного возврата результата. Вам обычно потребуется этот вариант только при работе со смешанной асинхронностью, когда группа входов будет обрабатываться частично синхронно, а частично асинхронно.

@spec ignore(t()) :: {:ok, term()} | {:exit, term()} | nil

Игнорирует существующую задачу.

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

Возвращает {:ok, reply} если ответ получен до игнорирования задачи, {:exit, reason} если задача завершилась аварийно до игнорирования, в противном случае nil.

Важно: избегайте использования Task.async/1,3, а затем немедленного игнорирования задачи. Если вы хотите запустить задачи, результаты которых вас не интересуют, используйте Task.Supervisor.start_child/2 вместо этого.

@spec shutdown(t(), timeout() | :brutal_kill) :: {:ok, term()} | {:exit, term()} | nil

Отсоединяет и останавливает задачу, а затем проверяет наличие ответа.

Возвращает {:ok, reply} если ответ получен во время остановки задачи, {:exit, reason} если задача завершилась аварийно, в противном случае nil. После остановки вы больше не можете ожидать или вызывать её.

Второй аргумент — это либо таймаут, либо :brutal_kill. В случае таймаута задаче отправляется сигнал выхода :shutdown, и если она не завершается в течение таймаута, она убивается. С :brutal_kill задача убивается сразу. В случае ненормального завершения задачи (возможно, убитой другим процессом), эта функция завершится по той же причине.

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

Если задача не связана с каким-либо процессом, например, задачи, запущенные с помощью Task.completed/1, мы проверяем наличие ответа или ошибки соответствующим образом, но без остановки процесса.

Если монитор задачи уже отключен или получено сообщение, и в очереди сообщений нет ожидающего ответа, эта функция вернёт {:exit, :noproc} в качестве результата, так как причина завершения не может быть определена.

@spec start((-> any())) :: {:ok, pid()}

Запускает задачу.

fun должна быть анонимной функцией без аргументов.

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

Если текущий узел завершается, узел завершится даже если задача не была завершена. По этой причине рекомендуется использовать Task.Supervisor.start_child/2 вместо этого, что позволяет контролировать время завершения через :shutdown параметр.

@spec start(module(), atom(), [term()]) :: {:ok, pid()}

Запускает задачу.

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

Если текущий узел завершается, узел завершится даже если задача не была завершена. По этой причине рекомендуется использовать Task.Supervisor.start_child/2 вместо этого, что позволяет контролировать время завершения через :shutdown параметр.

@spec start_link((-> any())) :: {:ok, pid()}

Запускает задачу в рамках дерева диспетчеризации с указанным fun.

fun должна быть анонимной функцией без аргументов.

Это используется для запуска статически управляемой задачи в дереве диспетчеризации.

@spec start_link(module(), atom(), [term()]) :: {:ok, pid()}

Запускает задачу в рамках дерева диспетчеризации с заданными module, function, и args.

Это используется для запуска статически управляемой задачи в дереве диспетчеризации.

@spec yield(t(), timeout()) :: {:ok, term()} | {:exit, term()} | nil

Временно блокирует процесс-отправителя, ожидая ответа задачи.

Возвращает {:ok, reply} если ответ получен, nil если ответ не получен, или {:exit, reason} если задача завершилась. Обратите внимание, что обычно неудача задачи также приводит к выходу процесса, которому принадлежит задача. Поэтому эта функция может вернуть {:exit, reason} если выполняется хотя бы одно из следующих условий:

• процесс задачи завершился по причине :normal
• процесс задачи не связан с отправителем (задача была запущена с помощью Task.Supervisor.async_nolink/2 или Task.Supervisor.async_nolink/4)
• отправитель обрабатывает завершения процессов

Таймаут, в миллисекундах или :infinity, может быть задан со значением по умолчанию 5000. Если время истекает до получения сообщения от задачи, эта функция вернёт nil и мониторинг останется активным. Следовательно, yield/2 может быть вызвано несколько раз для одной и той же задачи.

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

Если вы хотите завершить задачу, если она не ответит в течение timeout миллисекунд, вам нужно объединить эту функцию с shutdown/1, например:

case Task.yield(task, timeout) || Task.shutdown(task) do
  {:ok, result} ->
    result

  nil ->
    Logger.warning("Failed to get a result in #{timeout}ms")
    nil
end

Если вы хотите проверить задачу, но оставить её работающей после таймаута, вы можете объединить эту функцию с ignore/1, например:

case Task.yield(task, timeout) || Task.ignore(task) do
  {:ok, result} ->
    result

  nil ->
    Logger.warning("Failed to get a result in #{timeout}ms")
    nil
end

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

@spec yield_many([t()], timeout()) :: [{t(), {:ok, term()} | {:exit, term()} | nil}]

@spec yield_many([t()],
  timeout: timeout(),
  on_timeout: :nothing | :ignore | :kill_task
) :: [
  {t(), {:ok, term()} | {:exit, term()} | nil}
]

Вызывает множество задач в заданном интервале времени.

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

Аналогично yield/2, результат каждой задачи будет

• {:ok, term} если задача успешно вернула свой результат в заданном интервале времени
• {:exit, reason} если задача завершилась
• nil если задача продолжает выполняться после таймаута

См. yield/2 для получения дополнительной информации.

Пример

Task.yield_many/2 позволяет разработчикам запускать несколько задач и получать результаты, полученные в заданный интервал времени. Если объединить его с Task.shutdown/2 (или Task.ignore/1), то можно собрать эти результаты и отменить (или игнорировать) задачи, которые не ответили вовремя.

Давайте посмотрим пример.

tasks =
  for i <- 1..10 do
    Task.async(fn ->
      Process.sleep(i * 1000)
      i
    end)
  end

tasks_with_results = Task.yield_many(tasks, timeout: 5000)

results =
  Enum.map(tasks_with_results, fn {task, res} ->
    # Shut down the tasks that did not reply nor exit
    res || Task.shutdown(task, :brutal_kill)
  end)

# Here we are matching only on {:ok, value} and
# ignoring {:exit, _} (crashed tasks) and `nil` (no replies)
for {:ok, value} <- results do
  IO.inspect(value)
end

В примере выше создаются задачи, которые спят от 1 до 10 секунд и возвращают количество проспанных секунд. Если выполнить код сразу, вы увидите числа от 1 до 5, так как это задачи, которые ответили в заданный промежуток времени. Все остальные задачи будут закрыты с помощью вызова Task.shutdown/2.

Для удобства вы можете добиться аналогичного поведения, задав опцию :on_timeout как :kill_task (или :ignore). См. Task.await_many/2, если вы хотите выйти из вызывающего процесса при таймауте.

Опции

Вторым аргументом является либо таймаут, либо опции, которые по умолчанию:

• :timeout - максимальное время (в миллисекундах или :infinity) выполнения каждой задачи. По умолчанию 5000.

• :on_timeout - действие при таймауте задачи. Возможные значения:

  • :nothing - ничего не делать (по умолчанию). На задачи всё ещё можно ожидать, вызывать, игнорировать или завершать позже.
  • :ignore - результаты задачи будут проигнорированы.
  • :kill_task - задача, превысившая таймаут, будет убита.