Структура 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 - (с версии 1.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)

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

Внимание: неограниченный асинхронный + 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

Не рекомендуется запускать длительную задачу внутри поведения 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

Не рекомендуется запускать длительные задачи внутри поведения 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()],
  limit: pos_integer(),
  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, если нужно завершить вызывающий процесс при истечении времени ожидания.

Параметры

Второй аргумент — это либо таймаут, либо параметры, по умолчанию:

• :limit — максимальное количество задач для ожидания. Если предел достигнут до истечения таймаута, функция возвращает немедленно, не вызывая поведение :on_timeout

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

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

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