タスク の

タスクについて

タスクとは、ひとつの特定の操作を実行するためのプロセスです。 通常は他のプロセスとやり取りしませんが、タスクを開始したプロセスに結果を返すことはできます。

タスクは、作業を並列化するのによく使われます。

async/await

タスクを開始するには、Task.async/1を使います。引数として無名関数を受け取り、呼び出し元のプロセスにリンクされた新しいプロセスでそれを実行します。返り値は%Task{}構造体です。

実行結果を得るには、%Task{}構造体をTask.await/2に渡します。タスクが終了するのを待って、その結果を返します。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から始めることを意識しておくことが大切です。つまり、このように順番に使うと、タイムアウトが積み重なっていきます。

最初に待機されるタスクはタイムアウトを超えて実行されることはありませんが、2番目に待機されるタスクは、最初のタスクが実行された時間にタイムアウトを加えた時間だけ実行されます。

それぞれのタスクの実行時間が前のタスクより少しずつ長いものの、実行時間の差が待機のタイムアウトより小さい状況を考えてみましょう。

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