Завдання у

Про концепцію Завдання

Завдання - це процеси, призначені виконувати одну конкретну операцію. Зазвичай вони не спілкуються з іншими процесами, але можуть повернути результат процесу, який запустив завдання.

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

async/await

Щоб запустити завдання, використаймо Task.async/1. Вона приймає анонімну функцію як аргумент і виконує її в новому процесі, звʼязаному з процесом, який її викликав. Вона повертає структуру %Task{}.

Щоб отримати результат виконання, передаймо структуру %Task{} у Task.await/2. Вона дочекається завершення завдання й поверне його результат. Другий аргумент - це тайм-аут у мілісекундах, який типово становить 5000. Як тайм-аут можна передати значення :infinity, щоб завдання могло виконуватися без тайм-ауту.

task = Task.async(fn -> Enum.sum(0..999_999) end)
# => %Task{
#      owner: #PID<0.110.0>,
#      pid: #PID<0.119.0>,
#      ref: #Reference<0.3715954046.815529991.83334>
#    }

Task.await(task)
# => 499999500000

Зауважмо, що між запуском завдання й очікуванням на нього процес, який запустив завдання, не блокується й може виконувати інші операції.

Будь-яке завдання, запущене через Task.async/1, слід дочекатися, адже воно надсилає повідомлення процесу, який його запустив. Task.await/2 можна викликати для кожного завдання лише один раз.

start/start_link

Якщо потрібно запустити завдання лише заради побічних ефектів, скористаймося Task.start/1 або Task.start_link/1. Task.start/1 запускає завдання, не звʼязане з процесом, який його викликав, а Task.start_link/1 запускає завдання, звʼязане з процесом, який його викликав. Обидві функції повертають кортеж {:ok, pid}.

Тайм-аути

Поширений прийом - запустити асинхронні завдання, а потім дочекатися їх за допомогою Enum.map/2 ось так:

inputs
|> Enum.map(&Task.async(fn -> function.(&1) end))
|> Enum.map(&Task.await(&1, 1_000))

Важливо памʼятати, що Task.await синхронна, і щоразу, коли її викликають, відлік тайм-ауту починається з 0. Це означає, що коли її використовують у такий спосіб, послідовно, тайм-аути додаються.

Перше завдання, на яке очікують, виконуватиметься не довше за тайм-аут, а друге завдання, на яке очікують, матиме час, що дорівнює тривалості виконання першого завдання плюс тайм-аут.

Розгляньмо ситуацію, де кожне завдання виконується трохи довше за попереднє, але різниця в тривалості виконання менша за тайм-аут очікування.

inputs = 1..10
function = fn n -> :timer.sleep(n * 900) && n end

Можемо скористатися функцією Erlang :timer.tc/1, щоб виміряти, як довго виконуватимуться ці завдання, у мікросекундах.

:timer.tc(fn ->
  inputs
  |> Enum.map(&Task.async(fn -> function.(&1) end))
  |> Enum.map(&Task.await(&1, 1_000))
end)

# => {9004517, [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]}

9004517 мікросекунд - це 9 секунд, значно більше за заданий тайм-аут у 1 секунду (1000 мілісекунд).

Якщо це саме та поведінка, якої ми хочемо, чудово, берімося! Якщо ні, то альтернатива - скористатися або Task.async_stream/3, або Task.await_many/2.

Task.async_stream/3

Task.async_stream/3 дає змогу запустити ту саму функцію для багатьох різних вхідних даних, кожну в окремому завданні. Завдання звʼязані з поточним процесом, так само як і під час запуску через Task.async/1. Вона повертає потік.

Вона приймає тайм-аут, який застосовується до кожного завдання окремо. Розгляньмо ті самі вхідні дані й ту саму функцію, використані з Task.async_stream/3:

inputs
|> Task.async_stream(function, timeout: 1_000)
|> Enum.to_list()

# => ** (exit) exited in: Task.Supervised.stream(1000)
#       ** (EXIT) time out
#       (elixir 1.11.3) lib/task/supervised.ex:304: Task.Supervised.stream_reduce/7
#       (elixir 1.11.3) lib/enum.ex:3473: Enum.reverse/1
#       (elixir 1.11.3) lib/enum.ex:3066: Enum.to_list/1

Task.async_stream/3 чудово підходить, якщо потрібно запустити одну функцію з різними вхідними даними. Вона дає багато контролю завдяки іншим корисним опціям, як-от скільки завдань запускати одночасно і що робити в разі тайм-ауту завдання.

Якщо потрібно запускати завдання, які виконують різні операції, замість неї може знадобитися Task.await_many/2.

Task.await_many/2

Task.await_many/2 була додана в Elixir 1.11. Вона дає змогу дочекатися багатьох завдань зі спільним тайм-аутом. Розгляньмо ті самі вхідні дані й ту саму функцію, використані з Task.await_many/2:

inputs
|> Enum.map(&Task.async(fn -> function.(&1) end))
|> Task.await_many(1_000)

# => ** (exit) exited in: Task.await_many([...], 1000)
#        ** (EXIT) time out
#        (elixir 1.11.3) lib/task.ex:725: Task.await_many/5
#        (elixir 1.11.3) lib/task.ex:709: Task.await_many/2