TaskGroup으로 구조화된 동시성 설계하기

Python 3.11부터 — TaskGroup

이 장의 내용은 Python 3.11 이상에서만 동작합니다. 그 이전 버전을 써야 한다면 gather 로 돌아가거나 서드파티 라이브러리 anyio 의 TaskGroup 을 대신 써야 합니다. 사내 프로젝트가 아직 3.9, 3.10에 머물러 있다면 이번 장은 "언젠가 마이그레이션할 이유" 정도로 읽어도 됩니다.

앞 장에서 gather 의 문제를 봤습니다. 하나가 실패해도 나머지가 취소되지 않고, 취소하고 싶으면 직접 로직을 짜야 했습니다. asyncio.TaskGroup 은 이 문제를 언어 차원에서 풀기 위해 나온 기능입니다.

import asyncio

async def flaky():
    await asyncio.sleep(0.5)
    raise ValueError("네트워크 오류")

async def slow():
    try:
        await asyncio.sleep(3)
        return "느리지만 성공"
    except asyncio.CancelledError:
        print("slow() 취소됨")
        raise

async def main():
    async with asyncio.TaskGroup() as tg:
        tg.create_task(flaky())
        tg.create_task(slow())

asyncio.run(main())

flaky() 가 0.5초 뒤 예외를 던지면, TaskGroup 은 같은 그룹 안의 다른 태스크(slow())를 자동으로 취소합니다. 콘솔에 slow() 취소됨 이 곧바로 찍히는 걸 확인할 수 있습니다. gather 에서는 3초를 다 기다려야 끝났을 좀비 태스크가, TaskGroup 에서는 즉시 정리됩니다. 이게 "구조화된 동시성(structured concurrency)"이라는 개념의 핵심입니다 — async with 블록을 벗어나는 순간, 그 안에서 만든 모든 태스크의 생명주기가 반드시 끝나 있다는 게 보장됩니다. 블록 안에서 만든 태스크가 블록 밖으로 "새어나가는" 일이 없습니다.

여러 개가 동시에 실패하면? — ExceptionGroup

gather 는 첫 번째 예외만 던지고 끝입니다. 두 번째, 세 번째 태스크가 다른 이유로 실패했어도 그 정보는 사라집니다. TaskGroup 은 이 문제도 다르게 다룹니다.

async def fail_a():
    await asyncio.sleep(0.1)
    raise ValueError("A 실패")

async def fail_b():
    await asyncio.sleep(0.1)
    raise TypeError("B 실패")

async def main():
    async with asyncio.TaskGroup() as tg:
        tg.create_task(fail_a())
        tg.create_task(fail_b())

try:
    asyncio.run(main())
except* ValueError as eg:
    print("ValueError 그룹:", eg.exceptions)
except* TypeError as eg:
    print("TypeError 그룹:", eg.exceptions)

TaskGroup 은 발생한 예외들을 낱개로 버리지 않고 ExceptionGroup 하나로 묶어서 던집니다. except* (PEP 654, 이것도 3.11부터) 구문은 이 그룹을 예외 타입별로 걸러서 처리할 수 있게 해줍니다. 위 예제에서는 ValueError 그룹과 TypeError 그룹이 각각의 except* 절에서 따로 잡힙니다 — 기존 except 였다면 둘 중 먼저 도착한 예외 하나만 보고 나머지는 놓쳤을 상황입니다.

그렇다고 gather를 완전히 버릴 필요는 없다

개인적으로는 새 코드를 짤 때 gather 보다 TaskGroup 을 기본값으로 둡니다. "실패 시 나머지를 취소한다"는 동작이 대부분의 실전 상황(여러 API를 동시에 호출해 하나라도 실패하면 전체를 실패로 처리하는 경우)에 더 맞기 때문입니다. 다만 "일부가 실패해도 나머지 결과는 받고 싶다"는 요구라면 여전히 gather(..., return_exceptions=True) 가 더 간단합니다. TaskGroup 은 "전부 성공하거나, 실패하면 다 함께 정리한다"는 all-or-nothing 시나리오에 맞는 도구지, gather 의 완전한 대체재는 아닙니다.