Task Groups

Menjalankan banyak coroutine sekaligus mudah; menjaganya tetap rapi saat salah satu gagal adalah bagian yang sulit. Sebelum Python 3.11, pola asyncio.gather() menuntut kita menentukan sendiri apa yang terjadi pada task yang masih hidup ketika tetangganya gagal — dan jawaban yang salah berarti task bocor berjalan di latar belakang tanpa pengawasan. Class asyncio.TaskGroup (PEP 654) menyelesaikannya dengan satu aturan tegas: semua task yang lahir di dalam blok async with harus selesai bersama blok tersebut. Materi dasar pembuatan task telah dibahas pada Chapter Tasks dan Task Groups; Chapter ini membedah mekanisme kegagalannya dan membandingkannya dengan alat lama.

Rekap Cepat: Satu Blok untuk Semua Task #

TaskGroup adalah context manager async. Method create_task() menjadwalkan coroutine, dan keluar dari blok async with menunggu seluruh task selesai:

import asyncio


async def ambil(nama: str, jeda: float) -> str:
    await asyncio.sleep(jeda)
    return f"data {nama}"


async def main():
    async with asyncio.TaskGroup() as tg:
        t1 = tg.create_task(ambil("profil", 0.10))
        t2 = tg.create_task(ambil("pesanan", 0.20))
        t3 = tg.create_task(ambil("riwayat", 0.15))
    # keluar dari blok = semua task dijamin selesai
    print(t1.result())
    print(t2.result())
    print(t3.result())


asyncio.run(main())

Output yang Dihasilkan:

data profil
data pesanan
data riwayat

Meskipun task “riwayat” dan “profil” selesai lebih cepat, baris print() pertama baru berjalan setelah blok async with keluar — pada titik itu ketiga objek Task dijamin sudah done, sehingga t2.result() aman dipanggil langsung tanpa await.

Kegagalan Memicu Pembatalan Otomatis #

Perilaku andalan TaskGroup muncul saat sebuah task gagal. Alurnya: task yang gagal menandai grup, semua task lain langsung dibatalkan, grup menunggu pembatalan selesai, lalu seluruh kegagalan dibungkus menjadi satu ExceptionGroup dan dinaikkan dari statement async with:

import asyncio


async def tugas_sehat(nama: str) -> None:
    await asyncio.sleep(0.3)
    print(f"{nama} selesai")


async def tugas_gagal() -> None:
    await asyncio.sleep(0.05)
    raise ConnectionError("koneksi database putus")


async def main() -> None:
    async with asyncio.TaskGroup() as tg:
        tg.create_task(tugas_sehat("sinkronisasi"))
        tg.create_task(tugas_sehat("laporan"))
        tg.create_task(tugas_gagal())


try:
    asyncio.run(main())
except* ConnectionError as eg:
    print("Galat tertangkap:", [str(e) for e in eg.exceptions])

Output yang Dihasilkan:

Galat tertangkap: ['koneksi database putus']

Perhatikan apa yang tidak ada pada output: dua tugas_sehat tidak pernah mencetak "selesai". Keduanya tidur 0.3 detik ketika tugas_gagal melempar galat pada detik ke-0.05; TaskGroup membatalkan keduanya sebelum sempat selesai, sehingga satu-satunya baris output berasal dari handler except* di main level. Tidak ada task yang bocor, tidak ada Task was destroyed but it is pending.

Bentuk mentahnya bisa dilihat bila galat dibiarkan tidak tertangani. Simpan sebagai demo_taskgroup.py (versi tanpa blok try/except di sekeliling asyncio.run), lalu jalankan python3 demo_taskgroup.py:

Output yang Dihasilkan (dicetak interpreter Python ke stderr):

  + Exception Group Traceback (most recent call last):
  |   File "demo_taskgroup.py", line 20, in <module>
  |     asyncio.run(main())
  |   ... frame internal module asyncio diringkas ...
  |   File "demo_taskgroup.py", line 15, in main
  |     async with asyncio.TaskGroup() as tg:
  |                ^^^^^^^^^^^^^^^^
  |   ... frame internal module asyncio diringkas ...
  | ExceptionGroup: unhandled errors in a TaskGroup (1 sub-exception)
  +-+---------------- 1 ----------------
    | Traceback (most recent call last):
    |   File "demo_taskgroup.py", line 11, in tugas_gagal
    |     raise ConnectionError("koneksi database putus")
    | ConnectionError: koneksi database putus
    +------------------------------------

Dua penanda itu cerita utamanya: baris ExceptionGroup: unhandled errors in a TaskGroup menegaskan bahwa yang naik adalah grup buatan TaskGroup, dan blok bernomor 1 menunjukkan satu-satunya anak galat — ConnectionError dari tugas_gagal. Pembacaan traceback grup ini sendiri telah dibedah pada Chapter Exception Groups.

TaskGroup vs gather vs wait #

Perbedaan paling sering mengejutkan: gather(return_exceptions=True) menelan galat ke dalam daftar hasil. Dengan pola tersebut, galat berhenti menjadi exception dan menjadi nilai balasan biasa:

import asyncio


async def tugas_berhasil(nama: str) -> str:
    await asyncio.sleep(0.01)
    return f"{nama} selesai"


async def tugas_gagal() -> str:
    raise ValueError("data korup")


async def main() -> None:
    hasil = await asyncio.gather(
        tugas_berhasil("sinkronisasi"),
        tugas_gagal(),
        tugas_berhasil("laporan"),
        return_exceptions=True,
    )
    print("hasil gather:", hasil)


asyncio.run(main())

Output yang Dihasilkan:

hasil gather: ['sinkronisasi selesai', ValueError('data korup'), 'laporan selesai']

Pada output di atas, ValueError duduk tenang di antara hasil sukses — program tidak berhenti, dan tidak ada pembatalan pada task lain. Perilaku membiarkan-berjalan ini tepat untuk kasus seperti “kumpulkan semua respons yang bisa didapat”, namun berbahaya bila kegagalan seharusnya menghentikan seluruh operasi.

Prinsip Structured Concurrency #

TaskGroup membawa structured concurrency ke Python: setiap task hidup di dalam sebuah blok bersarang, sehingga masa hidup task selalu berada di dalam masa hidup blok pembentuknya — mirip dengan cara with mengikat masa hidup file ke sebuah blok. Konsekuensinya, tidak ada lagi task yatim yang berjalan setelah function pembentuknya kembali; debugger dan profiler pun bisa menunjukkan parent setiap task secara akurat. Bila skenario kita butuh task yang hidup lama lepas dari scope (misalnya background worker), TaskGroup memang bukan alatnya — penanganannya dibahas pada Chapter Queue dan Worker Patterns.