Tasks dan Task Groups

await pada satu coroutine tetap berarti menunggu satu hal. Kekuatan sesungguhnya async muncul ketika coroutine dijadwalkan sebagai task: unit kerja yang event loop jalankan bersamaan tanpa menunggu kita meng-await-nya satu per satu. Chapter ini membahas asyncio.create_task(), cara modern asyncio.TaskGroup (Python 3.11+), dan kapan memilih keduanya dibanding gather.

Membuat Task dengan create_task #

asyncio.create_task(coro) langsung menjadwalkan coroutine ke event loop dan mengembalikan object Task. Task mulai berjalan begitu loop mendapat kesempatan — bahkan sebelum kita menyentuh await:

import asyncio

async def ambil_data(nama):
    print(f"mulai mengambil data {nama}")
    await asyncio.sleep(1)
    print(f"selesai mengambil data {nama}")
    return f"data {nama}"

async def main():
    tugas = asyncio.create_task(ambil_data("Budi"))
    print("main melanjutkan pekerjaan lain")
    await asyncio.sleep(0.5)
    print("main siap menunggu hasil")
    hasil = await tugas
    print("hasil:", hasil)

asyncio.run(main())

Output yang Dihasilkan:

main melanjutkan pekerjaan lain
mulai mengambil data Budi
main siap menunggu hasil
selesai mengambil data Budi
hasil: data Budi

Mari kita telaah apa yang terjadi pada baris-baris kode di atas:

1. Baris 10 (tugas = asyncio.create_task(...)): coroutine ambil_data("Budi") dibungkus menjadi task dan masuk antrean event loop. Isinya belum berjalan karena main masih memegang kontrol.
2. Baris 11 (print("main melanjutkan...")): dicetak lebih dulu — bukti create_task tidak memblokir.
3. Baris 12 (await asyncio.sleep(0.5)): main menyerahkan kontrol; baru di sinilah event loop menjalankan task ambil_data, yang mencetak baris keduanya lalu tidur.
4. Baris 14 (hasil = await tugas): main menunggu task; karena task sudah berjalan setengah jalan, total tunggunya tinggal tersisa 0.5 detik, dan hasilnya diambil dari task tanpa menjalankan ulang.

Menunggu Task yang Sama Dua Kali #

Menyimpan task lalu meng-await-nya berulang kali tidak menjalankan ulang kerjanya. Task menyimpan hasil (atau exception) setelah selesai, dan await berikutnya hanya mengambil nilai yang tersimpan:

import asyncio
import time

async def ambil_harga():
    await asyncio.sleep(1)
    return 15_000

async def main():
    mulai = time.perf_counter()
    tugas = asyncio.create_task(ambil_harga())
    harga_pertama = await tugas
    harga_kedua = await tugas
    print(harga_pertama, harga_kedua)
    print(f"total {time.perf_counter() - mulai:.1f} detik")

asyncio.run(main())

Output yang Dihasilkan (nilai total waktu dapat sedikit berbeda di mesin Anda):

15000 15000
total 1.0 detik

Dua await menghasilkan 15000 berdua, tetapi total waktu tetap 1 detik — coroutine ambil_harga dijalankan tepat satu kali. Pola “mulai task, lanjut bekerja, baru ambil hasil” ini berguna untuk memanfaatkan waktu tunggu I/O dengan pekerjaan lain.

TaskGroup: Structured Concurrency #

Sejak Python 3.11, cara yang direkomendasikan untuk mengelola sekelompok task adalah asyncio.TaskGroup lewat async with. Semua task yang dibuat di dalam blok dijamin selesai (atau dibatalkan) sebelum blok ditinggalkan — prinsip ini disebut structured concurrency:

import asyncio

async def ambil_data(nama, detik):
    print(f"mulai mengambil data {nama}")
    await asyncio.sleep(detik)
    return f"data {nama}"

async def main():
    async with asyncio.TaskGroup() as grup:
        t1 = grup.create_task(ambil_data("Budi", 2))
        t2 = grup.create_task(ambil_data("Siti", 1))
    print(t1.result(), t2.result())

asyncio.run(main())

Output yang Dihasilkan:

mulai mengambil data Budi
mulai mengambil data Siti
data Budi data Siti

Baris print(t1.result(), t2.result()) baru dieksekusi setelah async with selesai — TaskGroup menunggu semua anggotanya tuntas lebih dulu. t1.result() dan t2.result() kemudian mengambil hasil tanpa await karena task dijamin sudah selesai.

Ketika Sebuah Task Gagal #

Keunggulan besar TaskGroup tampak saat ada yang error. Simpan kode berikut sebagai taskgroup_error.py:

import asyncio

async def ambil_data(nama, detik):
    await asyncio.sleep(detik)
    return f"data {nama}"

async def tugas_gagal():
    await asyncio.sleep(0.1)
    raise ValueError("server merespons 500")

async def main():
    async with asyncio.TaskGroup() as grup:
        grup.create_task(ambil_data("Budi", 2))
        grup.create_task(ambil_data("Siti", 2))
        grup.create_task(tugas_gagal())

asyncio.run(main())

Output yang Dihasilkan (jalankan sebagai python3 taskgroup_error.py; path pada traceback Anda dapat berbeda):

  + Exception Group Traceback (most recent call last):
  |   File "taskgroup_error.py", line 17, in <module>
  |     asyncio.run(main())
  |   File "/usr/lib/python3.12/asyncio/runners.py", line 195, in run
  |     return runner.run(main)
  |            ^^^^^^^^^^^^^^^^
  |   File "/usr/lib/python3.12/asyncio/runners.py", line 118, in run
  |     return self._loop.run_until_complete(task)
  |            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  |   File "/usr/lib/python3.12/asyncio/base_events.py", line 691, in run_until_complete
  |     return future.result()
  |            ^^^^^^^^^^^^^^^
  |   File "taskgroup_error.py", line 12, in main
  |     async with asyncio.TaskGroup() as grup:
  |                ^^^^^^^^^^^^^^^^^^^
  |   File "/usr/lib/python3.12/asyncio/taskgroups.py", line 71, in __aexit__
  |     return await self._aexit(et, exc)
  |            ^^^^^^^^^^^^^^^^^^^^^^^^^^
  |   File "/usr/lib/python3.12/asyncio/taskgroups.py", line 164, in _aexit
  |     raise BaseExceptionGroup(
  | ExceptionGroup: unhandled errors in a TaskGroup (1 sub-exception)
  +-+---------------- 1 ----------------
    | Traceback (most recent call last):
    |   File "taskgroup_error.py", line 9, in tugas_gagal
    |     raise ValueError("server merespons 500")
    | ValueError: server merespons 500
    +------------------------------------

Perhatikan tiga hal penting dari output tersebut:

1. Error ValueError dari tugas_gagal dibungkus menjadi ExceptionGroup — satu wadah untuk semua kegagalan di dalam grup.
2. Task ambil_data("Budi", 2) dan ambil_data("Siti", 2) tidak pernah menyelesaikan pekerjaannya — keduanya dibatalkan otomatis oleh TaskGroup begitu satu anggota gagal, sehingga tidak ada task yatim yang terus berjalan.
3. Seluruh async with menaikkan exception tersebut ke main(), dan program berhenti dengan pesan galat yang jelas.

Untuk menangani ExceptionGroup tanpa menghentikan program, Python 3.11 juga menyediakan except* yang memfilter sub-exception berdasarkan tipenya:

import asyncio

async def ambil_data(nama, detik):
    await asyncio.sleep(detik)
    return f"data {nama}"

async def tugas_gagal():
    await asyncio.sleep(0.1)
    raise ValueError("server merespons 500")

async def main():
    try:
        async with asyncio.TaskGroup() as grup:
            grup.create_task(ambil_data("Budi", 2))
            grup.create_task(tugas_gagal())
    except* ValueError as grup_error:
        for galat in grup_error.exceptions:
            print("tertangkap:", galat)

asyncio.run(main())

Output yang Dihasilkan:

tertangkap: server merespons 500

grup_error.exceptions berisi daftar sub-exception yang lolos filter ValueError, sehingga kita bisa menanganinya satu per satu.

gather vs TaskGroup #

Bagi kode baru yang menargetkan Python 3.11 ke atas, TaskGroup adalah pilihan default yang lebih aman. gather tetap berguna ketika kita hanya butuh “jalankan ini semua, kembalikan hasilnya berurutan” tanpa perlu membatalkan task lain.

Task yang dibatalkan otomatis oleh TaskGroup tadi berhenti lewat mekanisme CancelledError — topik yang layak dibedah sendiri pada Chapter Cancellation dan Timeouts.