Python asyncio.TaskGroup: ให้งานพี่น้องล้มและยกเลิกเป็นกลุ่ม

การสร้างหลาย task ด้วย asyncio.create_task() แล้วจำ await เองทุกตัวดูง่าย จนมี task หนึ่งล้มก่อนหรือ request ถูกยกเลิก จากนั้น task ที่ไม่มีเจ้าของอาจทำงานต่อ กิน connection หรือทิ้ง exception ไว้หลัง response จบ asyncio.TaskGroup ทำให้ lifetime ของงานลูกอยู่ใน block เดียวกัน

WHY: concurrency ต้องมีขอบเขต

เมื่อออกจาก async with TaskGroup() งานทุกตัวต้องเสร็จ ถ้างานหนึ่ง raise exception ที่ไม่ใช่ cancellation กลุ่มจะยกเลิกพี่น้องที่เหลือ รอ cleanup แล้วรวมข้อผิดพลาดเป็น ExceptionGroup รูปแบบนี้ทำให้ caller รู้ว่า subtree จบจริง ไม่ต้องเดาว่ายังมี background task หลุดอยู่หรือไม่

เหมาะกับงานที่เกิดจาก request เดียวกัน เช่นอ่าน profile, permissions และ feature flags พร้อมกัน ถ้าผลลัพธ์ทั้งหมดจำเป็นต่อคำตอบ ความล้มเหลวของหนึ่งงานควรหยุดงานอื่นเพื่อลดเวลารอและ resource usage

HOW: ทดสอบ happy path และ cancellation

ตัวอย่างนี้ตรวจทั้งการเก็บผลลัพธ์และการยกเลิก sibling เมื่ออีก task ล้ม:

import asyncio
import unittest


async def collect_values():
    async with asyncio.TaskGroup() as group:
        first = group.create_task(asyncio.sleep(0, result="profile"))
        second = group.create_task(asyncio.sleep(0, result="permissions"))
    return first.result(), second.result()


async def sibling_is_cancelled():
    started = asyncio.Event()
    cancelled = asyncio.Event()

    async def slow_job():
        started.set()
        try:
            await asyncio.sleep(10)
        except asyncio.CancelledError:
            cancelled.set()
            raise

    async def broken_job():
        await started.wait()
        raise ValueError("invalid payload")

    try:
        async with asyncio.TaskGroup() as group:
            group.create_task(slow_job())
            group.create_task(broken_job())
    except* ValueError:
        pass
    return cancelled.is_set()


class TaskGroupTests(unittest.TestCase):
    def test_collects_results(self):
        self.assertEqual(
            asyncio.run(collect_values()),
            ("profile", "permissions"),
        )

    def test_failure_cancels_sibling(self):
        self.assertTrue(asyncio.run(sibling_is_cancelled()))


if __name__ == "__main__":
    unittest.main()

จุดที่พลาดบ่อย

อย่ากลืน CancelledError โดยไม่ raise ต่อ เพราะ component ของ structured concurrency ใช้ cancellation ภายใน หากต้อง cleanup ให้ใช้ try/finally หรือ catch แล้ว raise เหมือนตัวอย่าง และอย่าสร้าง task นอกกลุ่มเพื่อหลบ lifetime—ถ้างานต้องอยู่ต่อจริงควรส่งให้ durable queue ที่มี owner ชัดเจน

TaskGroup ไม่ได้ทำให้งาน CPU-bound ขนานขึ้น event loop ยังเหมาะกับ I/O หากฟังก์ชันคำนวณหนักให้แยก process/thread อย่างระวัง นอกจากนี้ failure หลายตัวอาจอยู่ใน ExceptionGroup; handler ควรใช้ except* เฉพาะชนิดที่รับมือได้แล้วปล่อยชนิดอื่นขึ้นไป

Recap

ใช้ TaskGroup เมื่อ task หลายตัวเป็นครอบครัวเดียวกัน สร้างทั้งหมดใน block, ปล่อยให้กลุ่มรอและยกเลิกอย่างเป็นระบบ และทดสอบ cancellation path ด้วย งานถัดไปคือค้นหา create_task() ที่ไม่มีการเก็บ reference ใน request handler แล้วห่อ lifetime ให้ชัดครับ