Task

GenServer dan Supervisor pada topik-topik sebelumnya dirancang untuk process yang hidup lama dan menyimpan state, padahal banyak pekerjaan hanya berjalan sekali, misalnya mengolah satu berkas atau memanggil satu layanan. Task adalah abstraksi OTP untuk pekerjaan semacam itu, yaitu process sesaat yang dimulai, menjalankan satu fungsi, mengembalikan hasilnya, lalu berakhir. Topik ini membahas Task.async/1 dan Task.await/2, penularan kegagalan melalui link yang menyertainya, alternatif yang lebih aman dengan Task.Supervisor dan Task.yield/2, serta pengolahan koleksi paralel dengan Task.async_stream/3.

async dan await#

Task.async/1 memulai process baru yang langsung menjalankan fungsi yang diberikan, dan nilai balikannya berupa struct Task yang mewakili pekerjaan itu. Task.await/2 menunggu process tersebut selesai dan mengembalikan nilai fungsi yang dijalankannya, sehingga pola penggunaannya terasa seperti pemanggilan fungsi biasa yang dieksekusi di samping:

tugas = Task.async(fn ->
  Process.sleep(40)
  {"laporan.txt", 11}
end)

IO.inspect(Task.await(tugas), label: "await")

await: {"laporan.txt", 11}

Struct Task yang dikembalikan Task.async/1 bukan sekadar label, melainkan pegangan yang berisi pid process tugas dan referensi monitor milik pemanggil. Fungsi Task.await/2 menggunakan pegangan itu untuk menunggu, dan menetapkan batas waktu default lima detik. Apabila tugas belum selesai sampai batas itu, pemanggil menerima exit, sehingga batas waktu pun berlaku dua arah seperti call pada GenServer. Nilai batas yang berbeda dapat diberikan sebagai argumen kedua, misalnya Task.await(tugas, 2000).

Di balik kesederhanaan itu terdapat hubungan yang tidak ringan. Task.async/1 membuat link antara process pemanggil dan process tugas, sehingga kegagalan tugas menular langsung ke pemanggil sesuai perilaku link pada topik Link versus monitor. Tugas yang melempar exception membuat pemanggil ikut berhenti, dan hal itu berlaku meskipun pemanggil sedang tidak memanggil await/2 sama sekali. Contoh berikut memasang trap_exit terlebih dahulu agar exit dari tugas dapat diamati sebagai pesan, bukan sebagai kematian process:

# pelaporan crash bawaan Logger dimatikan agar keluaran skrip bersih
:logger.remove_handler(:default)

Process.flag(:trap_exit, true)
Task.async(fn -> raise "data rusak" end)

receive do
  {:EXIT, _pid, {%RuntimeError{message: pesan}, _jejak}} ->
    IO.puts("exit ditangkap: error runtime: #{pesan}")
after
  1000 -> IO.puts("waktu habis")
end

exit ditangkap: error runtime: data rusak

Tanpa trap_exit, exit yang sama menghentikan process pemanggil, dan inilah jebakan utama Task.async/1 yang digunakan tanpa pengaman:

Task.Supervisor dan Task.yield#

Task.Supervisor adalah supervisor khusus yang anak-anaknya berupa task, sehingga setiap tugas lahir di bawah pengawasan yang jelas. Task.Supervisor.async_nolink/3 memulai tugas di bawah supervisor tersebut tanpa menghubungkannya dengan pemanggil, sehingga kegagalan tugas tidak menular. Nilai balikannya tetap struct Task, tetapi pembacaan hasilnya berpindah ke Task.yield/2, yang tidak pernah melempar error dan selalu mengembalikan salah satu dari tiga bentuk:

Bentuk {:exit, alasan} membawa informasi alasan penghentian yang sama dengan pesan {:EXIT, ...} pada topik Link versus Monitor, yaitu tuple yang berisi struktur exception dan jejak tumpukan (stacktrace) ketika tugas melempar exception. Setelah yield/2 mengembalikan nil atau {:exit, _}, tugas yang masih berjalan perlu dihentikan menggunakan Task.shutdown/1 agar tidak terus berjalan tanpa pengawasan.

Pemilihan antara Task dan GenServer menjadi jelas dengan melihat masa hidup pekerjaannya. Task tepat untuk pekerjaan yang dimulai, selesai, lalu tidak digunakan lagi, karena tidak ada state yang perlu dijaga dan tidak ada pesan lanjutan yang ditunggu. Begitu pekerjaan membutuhkan state antar-permintaan, menerima lebih dari satu jenis pesan, atau hidup sepanjang aplikasi, GenServer kembali menjadi pilihan yang tepat. Tugas yang berulang dengan pola sama pun lebih layak ditulis sebagai GenServer daripada Task yang dimulai ulang terus-menerus.

Skrip utuh: pengolahan berkas paralel#

Skrip berikut mengolah lima berkas secara paralel dengan dua tingkat kerumitan sekaligus. Dua tugas pertama dijalankan di bawah Task.Supervisor melalui async_nolink/3, satu sehat dan satu sengaja gagal, dan keduanya dibaca dengan Task.yield/2. Sisanya diproses dengan Task.async_stream/3 yang membatasi jumlah pekerjaan serentak pada dua berkas:

# tugas.exs — Task untuk pekerjaan sesaat: async_nolink dengan Task.yield, lalu async_stream terurut

# laporan crash bawaan Logger dimatikan agar keluaran skrip deterministik
:logger.remove_handler(:default)

defmodule OlahBerkas do
  # simulasi pengolahan satu berkas: menunggu sebentar, lalu menghitung karakter namanya
  def satu(nama) do
    Process.sleep(40)
    {nama, String.length(nama)}
  end
end

defmodule Tunggu do
  def hingga(_kondisi, 0), do: :waktu_habis

  def hingga(kondisi, sisa_ms) when sisa_ms > 0 do
    if kondisi.() do
      :ok
    else
      Process.sleep(10)
      hingga(kondisi, sisa_ms - 10)
    end
  end
end

# tugas berjalan di bawah pengawas, sehingga kegagalannya tidak menular ke pemanggil
{:ok, _} = Task.Supervisor.start_link(name: :pengawas_tugas)

tugas_sehat =
  Task.Supervisor.async_nolink(:pengawas_tugas, fn -> OlahBerkas.satu("catatan.txt") end)

tugas_gagal = Task.Supervisor.async_nolink(:pengawas_tugas, fn -> raise "data rusak" end)

IO.inspect(Task.yield(tugas_sehat, 2000), label: "yield tugas sehat")

case Task.yield(tugas_gagal, 2000) do
  {:exit, {%RuntimeError{message: pesan}, _jejak}} ->
    IO.puts("yield tugas gagal: error runtime: #{pesan}")

  lainnya ->
    IO.puts("yield tugas gagal: #{inspect(lainnya)}")
end

# kelima berkas diolah paralel dengan batas dua pekerjaan serentak, hasilnya tetap terurut
berkas = ["laporan.txt", "catatan.txt", "suhu.csv", "log", "data.csv"]

hasil =
  berkas
  |> Task.async_stream(&OlahBerkas.satu/1, max_concurrency: 2, timeout: 2000)
  |> Enum.to_list()
  |> Enum.map(fn {:ok, pasangan} -> pasangan end)

Enum.each(hasil, fn {nama, jumlah} -> IO.puts("#{nama}: #{jumlah} karakter") end)
IO.puts("pemanggil tetap hidup: #{Process.alive?(self())}")

yield tugas sehat: {:ok, {"catatan.txt", 11}}
yield tugas gagal: error runtime: data rusak
laporan.txt: 11 karakter
catatan.txt: 11 karakter
suhu.csv: 8 karakter
log: 3 karakter
data.csv: 8 karakter
pemanggil tetap hidup: true

Keluaran mengonfirmasi kontrak async_nolink/3 di dua baris pertama. Tugas sehat menghasilkan {:ok, {"catatan.txt", 11}}, sedangkan tugas yang melempar exception menghasilkan {:exit, ...} yang alasannya dibaca dan dicetak, dan process pemanggil tetap hidup sampai baris terakhir. Bagian async_stream/3 mengembalikan satu pasangan untuk setiap berkas berbentuk {:ok, hasil}, dan Enum.to_list/1 menyusunnya kembali mengikuti urutan daftar masukan, bukan urutan selesainya, sehingga lima berkas tersebut dilaporkan dalam urutan yang sama persis dengan daftarnya.

Task.async_stream/3 juga menjaga sisi pengawasan. Setiap elemen diolah oleh process tugas tersendiri di bawah supervisor yang sama dengan tugas tunggal, opsi max_concurrency membatasi process yang hidup serentak, dan opsi timeout berlaku per elemen sehingga satu berkas lambat hanya menghasilkan {:exit, :timeout} untuk entri itu, bukan kegagalan seluruh aliran.

Ringkasan#

• Task.async/1 memulai process sekali jalan dan Task.await/2 mengembalikan hasilnya, tetapi keduanya membuat link yang menularkan kegagalan tugas ke pemanggil.
• Task.Supervisor.async_nolink/3 memulai tugas di bawah supervisor tanpa link ke pemanggil, dan Task.yield/2 mengembalikan {:ok, hasil}, {:exit, alasan}, atau nil tanpa melempar error.
• Task.async_stream/3 mengolah koleksi secara paralel dengan max_concurrency dan timeout per elemen, lalu hasilnya tersusun ulang mengikuti urutan masukan.
• Tugas yang masih berjalan setelah yield/2 sebaiknya dihentikan menggunakan Task.shutdown/1 agar tidak terus berjalan tanpa proses induk yang mengelolanya.

Referensi#

• Modul Task — Hexdocs Elixir — dokumentasi async, await, yield, dan async_stream beserta kontrak nilainya.
• Modul Task.Supervisor — Hexdocs Elixir — supervisor khusus task dan perbedaan async_nolink dengan async.
• Modul Process — Hexdocs Elixir — perilaku link yang menjadi dasar penularan kegagalan Task.async.