Задача

Удобства для запуска и ожидания задач.

Задачи — это процессы, предназначенные для выполнения одного определенного действия на протяжении всего их жизненного цикла, часто с малым или без взаимодействия с другими процессами. Наиболее распространённый случай использования задач — преобразование последовательного кода в конкурентный код путём асинхронного вычисления значения:

task = Task.async(fn -> do_some_work() end)
res  = do_some_other_work()
res + Task.await(task)

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

Помимо async/1 и await/2, задачи также могут запускаться как часть дерева управления и динамически запускаться на удалённых узлах. Мы рассмотрим все три сценария далее.

async и await

Одно из распространённых применений задач — преобразование последовательного кода в конкурентный код с помощью Task.async/1 , сохраняя при этом его семантику. При вызове будет создан новый процесс, связанный и отслеживаемый вызывающим процессом. После завершения действия задачи сообщение с результатом будет отправлено вызывающему процессу.

Task.await/2 используется для чтения сообщения, отправленного задачей.

При использовании async следует учитывать две важные вещи:

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

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

   Если этого не требуется, используйте Task.start/1 или рассмотрите возможность запуска задачи под Task.Supervisor с использованием async_nolink или start_child.

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

Наблюдаемые задачи

Также можно запустить задачу под управлением надзирателя. Это часто делается путем определения задачи в собственном модуле:

defmodule MyTask do
  use Task

  def start_link(arg) do
    Task.start_link(__MODULE__, :run, [arg])
  end

  def run(arg) do
    # ...
  end
end

А затем передать его надзирателю:

Supervisor.start_link([MyTask])

Поскольку эти задачи контролируются надзирателем, а не непосредственно связаны с вызывающим процессом, к ним нельзя обратиться с помощью await. Обратите внимание, что start_link/1, в отличие от async/1, возвращает {:ok, pid} (что ожидается от надзирателей).

Обратите внимание, что use Task определяет функцию child_spec/1, позволяющую поместить определённый модуль в дерево управления.

Сгенерированный child_spec/1 может быть настроен с помощью следующих параметров:

• :id — идентификатор спецификации дочернего элемента, по умолчанию текущий модуль
• :start — способ запуска дочернего процесса (по умолчанию вызов __MODULE__.start_link/1)
• :restart — время перезапуска дочернего элемента, по умолчанию :temporary
• :shutdown — способ завершения дочернего элемента

В отличие от GenServer, Agent и Supervisor, задача имеет значение по умолчанию :restart :temporary. Это означает, что задача не будет перезапущена даже в случае аварийного завершения. Если вы хотите, чтобы задача перезапускалась при неудачном завершении, сделайте так:

use Task, restart: :transient

Если вы хотите, чтобы задача всегда перезапускалась:

use Task, restart: :permanent

См. документацию Supervisor для получения дополнительной информации.

Динамически контролируемые задачи

Модуль Task.Supervisor позволяет разработчикам динамически создавать несколько контролируемых задач.

Краткое описание:

{:ok, pid} = Task.Supervisor.start_link()
task = Task.Supervisor.async(pid, fn ->
  # Do something
end)
Task.await(task)

Однако в большинстве случаев вы хотите добавить надзирателя задач в своё дерево управления:

Supervisor.start_link([
  {Task.Supervisor, name: MyApp.TaskSupervisor}
])

Теперь вы можете динамически запускать контролируемые задачи:

Task.Supervisor.start_child(MyApp.TaskSupervisor, fn ->
  # Do something
end)

Или даже использовать шаблон async/await:

Task.Supervisor.async(MyApp.TaskSupervisor, fn ->
  # Do something
end) |> Task.await()

Наконец, см. Task.Supervisor для других поддерживаемых операций.

Распределённые задачи

Поскольку Elixir предоставляет надзирателя задач, легко использовать его для динамического запуска задач на разных узлах:

# On the remote node
Task.Supervisor.start_link(name: MyApp.DistSupervisor)

# On the client
Task.Supervisor.async({MyApp.DistSupervisor, :remote@local},
                      MyMod, :my_fun, [arg1, arg2, arg3])

Обратите внимание, что при работе с распределёнными задачами следует использовать функцию Task.Supervisor.async/4, которая ожидает явного модуля, функции и аргументов, вместо Task.Supervisor.async/2, которая работает с анонимными функциями. Это связано с тем, что анонимные функции требуют наличия той же версии модуля на всех участвующих узлах. См. документацию модуля Agent для получения дополнительной информации о распределённых процессах, поскольку ограничения, описанные там, применимы ко всей экосистеме.

Резюме

Типы

t()

Функции

%Task{}

Структура задачи

async(fun)

Запускает задачу, к которой нужно обратиться с помощью await

async(mod, fun, args)

Запускает задачу, к которой нужно обратиться с помощью await

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

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

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

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

await(task, timeout \\ 5000)

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

shutdown(task, shutdown \\ 5000)

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

start(fun)

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

start(mod, fun, args)

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

start_link(fun)

Запускает процесс, связанный с текущим процессом

start_link(mod, fun, args)

Запускает задачу как часть дерева управления

yield(task, timeout \\ 5000)

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

yield_many(tasks, timeout \\ 5000)

Возвращает множеству задач в заданный промежуток времени

Типы

t()

t() :: %Task{owner: term(), pid: term(), ref: term()}

Функции

%Task{} (структура)

Структура задачи.

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

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

• :ref — ссылка на монитор задачи

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

async(fun)

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

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

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

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

См. также async/3.

async(mod, fun, args)

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

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

Возвращается структура Task, содержащая соответствующую информацию. Разработчики должны в конечном итоге вызвать Task.await/2 или Task.yield/2, а затем Task.shutdown/2 на возвращенной задаче.

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

Связывание

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

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 так же, как вы бы это сделали, если бы не было вызова async. Например, чтобы вернуть {:ok, val} | :error результаты или, в более крайних случаях, с использованием try/rescue. Другими словами, асинхронная задача должна рассматриваться как расширение процесса, а не как механизм изоляции его от всех ошибок.

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

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

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

  Кроме того, даже при перехвате завершений вызов await все равно завершит процесс, если задача завершилась без отправки результата.

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

Формат сообщения

Ответ, отправленный задачей, будет иметь формат {ref, result}, где ref — ссылка на монитор, хранящаяся в структуре задачи, а result — возвращаемое значение функции задачи.

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

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

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

Каждый элемент 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 для обсуждения, опций и дополнительных примеров.

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

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

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

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

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

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

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

Опции

• :max_concurrency — устанавливает максимальное количество задач, выполняемых одновременно. По умолчанию равно System.schedulers_online/0.
• :ordered — определяет, должны ли результаты возвращаться в том же порядке, что и входной поток. Эта опция полезна при работе с большими потоками и позволяет не буферизировать результаты перед их передачей. По умолчанию true.
• :timeout — максимальное время (в миллисекундах), которое каждая задача может выполнять. По умолчанию 5000.

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

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

Пример

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

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)

await(task, timeout \\ 5000)

await(t(), timeout()) :: term() | no_return()

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

Таймаут в миллисекундах можно задать, значение по умолчанию 5000. В случае завершения процесса задачи, эта функция завершится по той же причине, что и задача.

Если таймаут истекает, await завершит работу; тем не менее, задача будет продолжать выполняться. При завершении вызывающего процесса его сигнал завершения прервет задачу, если она не перехватывает завершения.

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

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

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

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

Примеры

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

shutdown(task, shutdown \\ 5000)

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

Отключает и завершает задачу, а затем проверяет ответ.

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

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

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

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

start(fun)

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

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

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

start(mod, fun, args)

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

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

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

start_link(fun)

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

Запускает процесс, связанный с текущим процессом.

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

start_link(mod, fun, args)

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

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

yield(task, timeout \\ 5000)

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

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

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

• задача завершилась с причиной :normal
• она не связана с вызывающим процессом
• вызывающий процесс перехватывает завершения

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

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

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

case Task.yield(task, timeout) || Task.shutdown(task) do
  {:ok, result} ->
    result
  nil ->
    Logger.warn "Failed to get a result in #{timeout}ms"
    nil
end

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

yield_many(tasks, timeout \\ 5000)

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

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

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

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

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

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

Пример

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

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

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

tasks_with_results = Task.yield_many(tasks, 5000)

results = Enum.map(tasks_with_results, fn {task, res} ->
  # Shutdown 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.