파이썬 asyncio - 이벤트 루프와 async await, 블로킹의 대가 (파이썬 고급 19단원)
이 단원에서 배우는 것
18단원에서 CPU 를 태우는 로그 파싱은 프로세스로, 무언가를 기다리는 일은 스레드로 병렬화했다. 그런데 logkit 에 기능이 하나 붙는다. 로그에서 뽑아낸 엔드포인트 2,000개에 헬스체크를 보내 응답 시간을 재는 것이다. 스레드 2,000개는 만들 수는 있지만 메모리와 컨텍스트 전환 비용이 크고, 풀 크기를 100 으로 줄이면 대기 시간이 20배가 된다. 이 지점이 asyncio 의 자리다.
- 이벤트 루프가 하나의 스레드에서 수천 개 대기를 처리하는 원리를 설명한다
TaskGroup·timeout·Semaphore로 실패와 동시성을 통제한다 (3.11 신규 API)- 블로킹 코드가 섞였을 때 무슨 일이 벌어지는지 측정하고,
to_thread로 빼낸다
버전 기준은 파이썬 3.11 이상이다. 이 단원에서 쓰는 asyncio.TaskGroup, asyncio.timeout, except* 는 전부 3.11 에서 들어왔고, 이전 방식(gather, wait_for)보다 명확히 낫다.
왜 필요한가
스레드 1개가 HTTP 응답을 기다리는 동안 실제로 하는 일은 아무것도 없다. 그런데 OS 스레드는 스택을 잡고, 커널이 스케줄링 대상으로 관리하며, 전환마다 비용을 낸다. 대기가 전부인 작업 2,000개를 위해 OS 자원 2,000개를 쓰는 것은 낭비다.
asyncio 는 이 낭비를 없앤다. 스레드 하나가 "지금 기다리는 중"인 작업 수천 개의 목록을 들고 있다가, OS 에게 "이 소켓들 중 아무거나 준비되면 알려 달라"(epoll/kqueue)고 맡긴다. 준비된 것부터 이어서 실행한다. 작업 하나의 비용은 스택이 아니라 코루틴 객체 하나, 수 KB 수준이다.
대신 대가가 있다. 이 모델은 모든 작업이 자발적으로 제어권을 내놓는다는 전제 위에 서 있다. 한 작업이 CPU 를 붙잡고 놓지 않으면 나머지 1,999개가 전부 멈춘다. 스레드에서는 OS 가 강제로 뺏어 오지만 여기서는 아무도 뺏지 못한다. asyncio 의 어려움은 문법이 아니라 이 전제를 지키는 데 있다.
문법과 예제
코루틴은 호출해도 실행되지 않는다
여기서 대부분의 첫 혼란이 생긴다.
import asyncio
async def check(path: str) -> tuple[str, int]:
await asyncio.sleep(0.2) # 실제로는 HTTP 요청
return path, 200
async def main() -> None:
r = check("/api/orders")
print(type(r).__name__) # coroutine — 아직 아무것도 안 했다
print(await r) # 여기서 비로소 실행된다
asyncio.run(main())async def 함수를 호출하면 실행 계획서(코루틴 객체)만 돌아온다. 이것을 실행하는 방법은 두 가지다. await 로 지금 기다리거나, create_task 로 루프에 등록해 두고 나중에 결과를 받거나. await 만 늘어놓으면 순차 실행이 되고 아무것도 빨라지지 않는다.
await check("/a"); await check("/b") # 0.4초. 동시성이 아니다
await asyncio.gather(check("/a"), check("/b")) # 0.2초TaskGroup — 3.11 이후의 기본형
import asyncio
import time
PATHS = ["/api/orders", "/api/users", "/api/pay", "/api/stats"]
async def check(path: str) -> tuple[str, int]:
await asyncio.sleep(0.2)
if path == "/api/pay":
raise TimeoutError(path)
return path, 200
async def main() -> None:
started = time.perf_counter()
try:
async with asyncio.TaskGroup() as tg:
tasks = [tg.create_task(check(p)) for p in PATHS]
except* TimeoutError as eg:
print("실패:", [str(e) for e in eg.exceptions])
print("%.2fs" % (time.perf_counter() - started))
for t in tasks:
if not t.cancelled() and t.exception() is None:
print("성공:", t.result())
asyncio.run(main())TaskGroup 의 계약은 명확하다.
async with블록을 벗어날 때 등록된 모든 작업이 끝났음이 보장된다. 뒷정리를 잊을 수 없다.- 한 작업이 실패하면 나머지를 자동으로 취소한다. 실패한 배포에 요청을 계속 던지는 일이 없다.
- 여러 작업이 동시에 실패하면
ExceptionGroup으로 묶여 오고except*로 타입별로 받는다.gather는 첫 예외만 올리고 나머지는 버렸다.
취소가 실제로 일어나는지 확인해 보자.
async def fail_fast() -> None:
await asyncio.sleep(0.1)
raise TimeoutError("/api/pay")
async def slow(name: str) -> str:
await asyncio.sleep(5)
return name
async def main() -> None:
started = time.perf_counter()
try:
async with asyncio.TaskGroup() as tg:
tg.create_task(fail_fast())
long = tg.create_task(slow("/api/stats"))
except* TimeoutError as eg:
print("실패:", [str(e) for e in eg.exceptions])
print("취소됨:", long.cancelled(), "%.2fs" % (time.perf_counter() - started))
asyncio.run(main())실패: ['/api/pay']
취소됨: True 0.10s5초짜리 작업이 0.1초 만에 정리됐다. gather 였다면 5초를 채우거나, 취소를 직접 코딩해야 했다.
시간 예산과 동시성 제한
async def main() -> None:
# 3.11 신규. wait_for 보다 읽기 쉽고 블록 단위로 걸린다
try:
async with asyncio.timeout(0.1):
await check("/api/orders")
except TimeoutError:
print("0.1초 예산 초과")
# 상대 서버를 보호한다. 2,000개를 한 번에 던지면 상대가 죽는다
sem = asyncio.Semaphore(20)
async def limited(path: str):
async with sem:
return await check(path)
async with asyncio.TaskGroup() as tg:
tasks = [tg.create_task(limited(p)) for p in PATHS]
print([t.result() for t in tasks])Semaphore 없이 TaskGroup 에 2,000개를 넣으면 2,000개가 즉시 출발한다. 이건 동시성 제어가 아니라 부하 테스트다. 실무에서 asyncio 코드를 쓸 때 세마포어는 선택이 아니라 기본이다.
logkit 헬스체커 완성형
from __future__ import annotations
import asyncio
import time
from dataclasses import dataclass
@dataclass(frozen=True, slots=True)
class Health:
path: str
status: int | None
ms: float
error: str | None = None
async def probe(path: str, sem: asyncio.Semaphore, budget: float = 3.0) -> Health:
"""실패해도 예외를 밖으로 내지 않고 결과 객체로 만든다."""
started = time.perf_counter()
try:
async with sem, asyncio.timeout(budget):
status = await fetch_status(path) # 아래에서 정의
except TimeoutError:
return Health(path, None, (time.perf_counter() - started) * 1000, "timeout")
except Exception as e:
return Health(path, None, (time.perf_counter() - started) * 1000, repr(e))
return Health(path, status, (time.perf_counter() - started) * 1000)
async def check_all(paths: list[str], concurrency: int = 20) -> list[Health]:
sem = asyncio.Semaphore(concurrency)
async with asyncio.TaskGroup() as tg:
tasks = [tg.create_task(probe(p, sem)) for p in paths]
return [t.result() for t in tasks]설계 판단이 하나 들어 있다. 개별 실패를 예외로 올리지 않고 Health 객체로 만든다. 2,000개 중 3개가 죽었다고 전체를 취소하면 안 되기 때문이다. TaskGroup 의 "하나 실패하면 전부 취소"는 강력하지만, 그게 원하는 동작이 아닐 때는 각 작업이 스스로 실패를 흡수해야 한다. 반대로 하나라도 실패하면 전체가 무의미한 작업(트랜잭션의 여러 단계)에서는 예외를 그대로 올려 TaskGroup 이 취소하게 두는 편이 맞다.
fetch_status 는 실제로는 httpx 나 aiohttp 같은 async 지원 클라이언트로 구현한다. 표준 라이브러리만 쓸 거면 아래처럼 소켓 수준으로 붙일 수 있다.
async def fetch_status(url_path: str, host: str = "127.0.0.1", port: int = 8000) -> int:
reader, writer = await asyncio.open_connection(host, port)
try:
writer.write(f"HEAD {url_path} HTTP/1.0\r\nHost: {host}\r\n\r\n".encode())
await writer.drain()
first = await reader.readline()
return int(first.split()[1])
finally:
writer.close()
await writer.wait_closed()블로킹 코드가 섞이면 실제로 무슨 일이 생기는가
말로 하면 와닿지 않으니 재 보자. 0.05초마다 뛰는 하트비트를 심어 두고, 그 옆에서 동기 CPU 작업을 돌린다.
import asyncio
import time
async def heartbeat(stop: asyncio.Event) -> int:
n = 0
while not stop.is_set():
await asyncio.sleep(0.05)
n += 1
return n
def parse_big_file() -> int: # 동기 CPU 작업
return sum(i * i for i in range(8_000_000))
async def bad() -> None:
stop = asyncio.Event()
hb = asyncio.create_task(heartbeat(stop))
await asyncio.sleep(0.05)
started = time.perf_counter()
parse_big_file() # 이벤트 루프가 여기서 멈춘다
took = time.perf_counter() - started
stop.set()
print("블로킹 %.2fs 동안 하트비트 %d 회" % (took, await hb))
async def good() -> None:
stop = asyncio.Event()
hb = asyncio.create_task(heartbeat(stop))
await asyncio.sleep(0.05)
started = time.perf_counter()
await asyncio.to_thread(parse_big_file) # 스레드로 내보낸다
took = time.perf_counter() - started
stop.set()
print("to_thread %.2fs 동안 하트비트 %d 회" % (took, await hb))
asyncio.run(bad())
asyncio.run(good())블로킹 0.41s 동안 하트비트 1 회
to_thread 0.43s 동안 하트비트 8 회같은 0.4초인데 앞은 하트비트가 1회, 뒤는 8회다. 0.4초 동안 이벤트 루프 전체가 정지했다는 뜻이다. 웹 서버였다면 그 사이 들어온 모든 요청이 0.4초씩 늦어진다. 동시 요청 100개면 지연이 그대로 쌓인다. asyncio 앱의 p99 응답 시간이 튀는 원인의 대부분이 이것이다.
주의할 점은 to_thread 가 작업을 빠르게 만들지 않는다는 것이다. 0.41초가 0.43초로 오히려 늘었다. GIL 때문에 CPU 작업 자체는 여전히 직렬이고, 다만 5ms 마다 GIL 이 넘어가면서 이벤트 루프가 숨을 쉴 뿐이다. 진짜 CPU 작업이라면 18단원의 프로세스풀로 보내야 한다.
from concurrent.futures import ProcessPoolExecutor
pool = ProcessPoolExecutor()
async def parse_in_process(path: str) -> tuple[str, int, int]:
loop = asyncio.get_running_loop()
return await loop.run_in_executor(pool, file_stats, path) # 18단원의 함수블로킹의 범인은 CPU 만이 아니다. time.sleep, requests.get, 동기 DB 드라이버, open().read() 로 큰 파일 읽기, subprocess.run 전부 같은 결과를 낸다. async 함수 안에서 await 없이 호출되는 I/O 는 전부 의심 대상이다.
개발 중에는
asyncio.run(main(), debug=True)를 켜 두면 100ms 이상 루프를 잡은 콜백을 경고로 찍어 준다. 범인을 손으로 찾을 필요가 없다.
취소 보호(shield)와 큐, 동기화 프리미티브 전체 목록은 asyncio 태스크 공식 문서에 정리돼 있다.
실무에서 자주 틀리는 것
1. create_task 결과를 변수에 안 담아 작업이 사라진다
async def main():
for path in paths:
asyncio.create_task(probe(path)) # 반환값을 버린다
await asyncio.sleep(10)이벤트 루프는 실행 중인 태스크를 약한 참조로만 들고 있다. 강한 참조가 어디에도 없으면 가비지 컬렉터가 도중에 회수해 갈 수 있고, 작업은 아무 로그 없이 중간에 사라진다. 100개 중 97개만 처리되는 재현 안 되는 버그가 이렇게 만들어진다. 답은 TaskGroup 을 쓰는 것이다. 굳이 fire-and-forget 이 필요하다면 집합에 담아 두고 완료 시 제거한다.
background: set[asyncio.Task] = set()
def spawn(coro) -> None:
task = asyncio.create_task(coro)
background.add(task)
task.add_done_callback(background.discard)2. gather 의 예외 처리를 오해한다
results = await asyncio.gather(bad(), ok())기본 동작은 첫 예외를 즉시 올리고, 나머지 작업은 취소하지 않고 그대로 두는 것이다. 취소되지 않은 작업들은 계속 돌다가 아무도 결과를 안 가져가는 상태가 되고, 그중 하나가 또 실패하면 "Task exception was never retrieved" 경고만 남는다. return_exceptions=True 를 주면 예외가 결과 리스트에 값으로 들어오지만, 이번엔 결과를 검사하지 않으면 실패를 놓친다.
results = await asyncio.gather(*coros, return_exceptions=True)
failed = [r for r in results if isinstance(r, BaseException)]
if failed:
raise ExceptionGroup("헬스체크 실패", failed)새 코드라면 이런 고민이 필요 없는 TaskGroup 을 쓴다.
3. 동기 함수 안에서 asyncio.run 을 부른다
async def handler():
data = load_config() # 이 함수 안에서 asyncio.run(...) 을 부른다
RuntimeError: asyncio.run() cannot be called from a running event loopasyncio.run 은 새 이벤트 루프를 만들고 끝나면 닫는다. 이미 루프가 도는 안에서는 쓸 수 없다. 프로그램 전체에서 딱 한 번, 최상단 진입점에서만 부르는 것이 원칙이다. 라이브러리 코드에서는 절대 부르지 않는다. 동기 코드와 async 코드가 섞이기 시작하면 코드베이스 전체가 async 로 물드는데("function color" 문제), 이를 줄이려면 경계를 한 곳으로 몰아야 한다. 보통은 CLI 진입점이나 웹 프레임워크의 핸들러가 그 경계다.
4. 취소를 except Exception 으로 삼킨다
async def probe(path):
try:
return await fetch_status(path)
except Exception: # CancelledError 는 안 잡힌다 — 3.8 부터
return None3.8 부터 CancelledError 는 BaseException 을 상속하므로 위 코드는 다행히 취소를 삼키지 않는다. 문제는 except BaseException 이나 except asyncio.CancelledError: pass 를 쓰는 경우다. 취소를 삼키면 TaskGroup 이 종료를 기다리며 영원히 멈추고, 서비스 종료가 안 된다. 취소 시 정리할 것이 있다면 잡되 반드시 다시 던진다.
try:
return await fetch_status(path)
except asyncio.CancelledError:
await writer_cleanup()
raise # 필수5. 잘못된 도구를 async 로 감싼다
async def fetch(url):
return requests.get(url).json() # async 를 붙여도 블로킹은 블로킹이다async def 는 마법이 아니다. 안에 await 가 없으면 그냥 느린 동기 함수이며, 루프를 통째로 세운다. 동기 라이브러리밖에 없다면 선택지는 두 가지뿐이다. asyncio.to_thread 로 감싸거나, asyncio 를 포기하고 스레드풀로 가는 것이다. 후자가 부끄러운 선택이 아니다. 동시 연결이 수백 개 수준이면 스레드풀이 더 단순하고 디버깅도 쉽다.
스스로 확인하기
- 아래 코드는 2,000개 엔드포인트를 점검하는데 전체가 400초 걸린다. 개별 요청은 0.2초다. 무엇이 잘못됐고 어떻게 고치는가?
async def check_all(paths): results = [] for p in paths: results.append(await probe(p)) return results asyncio.timeout(3)으로 감싼 블록 안에서to_thread(parse_big_file)를 부르고, 이 파싱이 10초 걸린다. 3초 뒤에 무슨 일이 일어나는가?- 헬스체크 결과를 DB 에 넣으려는데 드라이버가 동기(psycopg2)뿐이다. 2,000건을 어떻게 처리하겠는가?
정답
await를 루프 안에서 하나씩 부르므로 완전한 순차 실행이다. 2,000 × 0.2 = 400초, 정확히 순차 실행 시간이다.async를 붙였다고 저절로 동시에 도는 것이 아니다.TaskGroup으로 전부 등록한 뒤 한꺼번에 기다리고, 세마포어로 동시 수를 제한한다.async def check_all(paths, concurrency=50): sem = asyncio.Semaphore(concurrency) async with asyncio.TaskGroup() as tg: tasks = [tg.create_task(probe(p, sem)) for p in paths] return [t.result() for t in tasks]- 3초 뒤
TimeoutError가 호출한 쪽에서 발생하지만, 스레드에서 도는parse_big_file은 멈추지 않는다. 파이썬은 실행 중인 스레드를 강제로 죽일 수 없다. 그 스레드는 10초를 채우고 나서야 풀에 반납되며, 그동안 스레드풀 슬롯 하나를 계속 점유한다. 이런 패턴을 반복하면 기본 스레드풀이 고갈되어 이후to_thread가 전부 대기에 걸린다. 취소 가능해야 하는 긴 작업은 스레드가 아니라 별도 프로세스에 두고 프로세스를 종료시키는 방식이어야 한다. - 2,000번의
to_thread는 스레드풀을 고갈시키고 커넥션도 그만큼 필요해진다. 결과를 모아서 한 번에 쓴다. 헬스체크가 전부 끝난 뒤executemany또는COPY로 한 번의to_thread안에서 벌크 삽입한다. 실시간성이 필요하면asyncio.Queue에 결과를 넣고, 100건 또는 1초마다 배치로 비우는 소비자 태스크 하나를 둔다. 어느 쪽이든 I/O 경계를 넘는 횟수를 줄이는 것이 핵심이고, 이건 asyncio 문제가 아니라 설계 문제다.
다음 단원에서는 방향을 바꿔, 지금까지 계속 써 온 @dataclass 같은 데코레이터와 프레임워크가 내부에서 무엇을 하는지 연다. 디스크립터와 메타클래스다.