큐를 즉시 종료해도 처리 중인 일은 남는다: asyncio.Queue.shutdown의 실제 계약

1. 종료 신호와 작업 완료는 같은 말이 아니다

오래 실행되는 asyncio 작업자 풀을 멈출 때 흔히 센티널 값을 큐에 넣는다. 소비자는 그 값을 꺼내면 반복문을 끝낸다. 작업자가 하나면 버틸 만하지만, 작업자가 늘어나면 센티널 개수와 종료 순서가 프로토콜의 일부가 된다. 생산자가 큐가 꽬 때 put()에서 막혀 있다면 센티널만으로는 생산자까지 깨우기도 어렵다.

Python 3.13에 추가된 asyncio.Queue.shutdown()은 이 문제를 큐 API 안으로 가져왔다. 호출 뒤에는 새 put()이 QueueShutDown을 내고, 이미 put()에서 기다리던 코루틴도 깨어난다. 큐가 종료됐고 비어 있으면 get()도 같은 예외를 낸다. 종료를 알리기 위해 업무 데이터와 센티널을 섞지 않아도 된다.

그렇다고 shutdown()이 작업자 생명주기까지 끝내 주는 것은 아니다. 특히 join()의 의미를 잘못 읽으면 "큐가 끝났다"를 "모든 부작용이 완료됐다"로 오해하기 쉽다. 핵심은 qsize()가 아니라 큐 내부의 미완료 작업 카운터다.

그림 1. join은 큐 길이가 아니라 put과 task_done이 만든 미완료 작업 카운터가 0이 될 때 풀린다. 출처: docs.python.org/3/library/asyncio-queue.html.

2. join()은 대기열 길이가 아니라 완료 통지를 센다

put()으로 항목 하나가 들어갈 때마다 미완료 카운터가 1 증가한다. 소비자가 get()으로 항목을 꺼내도 이 숫자는 줄지 않는다. 해당 항목의 처리가 끝난 뒤 task_done()을 호출해야 1이 감소한다. 카운터가 0이 되는 순간 join()이 풀린다.

이 구분은 실무에서 중요하다. 소비자가 항목을 꺼낸 직후 큐 길이는 0이 될 수 있지만, 데이터베이스 쓰기나 파일 업로드는 아직 진행 중일 수 있다. q.empty()가 참이라는 사실은 "가져갈 항목이 없다"는 뜻이지 "모든 일이 끝났다"는 뜻이 아니다.

반대로 task_done()을 너무 많이 호출하면 ValueError가 난다. 보통 소비자 루프에서 다음처럼 finally에 두되, 실제로 get()에 성공한 항목에 대해서만 실행하는 편이 안전하다.

async def worker(queue):
    while True:
        try:
            item = await queue.get()
        except asyncio.QueueShutDown:
            return
        try:
            await handle(item)
        finally:
            queue.task_done()

3. 기본 종료는 남은 일을 끝까지 처리한다

queue.shutdown(immediate=False)가 기본 동작이다. 이 모드에서는 새 항목을 더 넣을 수 없지만 이미 큐에 들어간 항목은 계속 get()할 수 있다. 소비자는 남은 항목을 처리하고 각각 task_done()을 호출한다. 큐가 비고 미완료 카운터까지 0이 되면 join()이 정상적으로 반환한다. 이후 get()은 QueueShutDown을 낸다.

정상 종료가 필요한 서비스라면 이 흐름이 먼저다. 요청 수신을 닫고, 생산자가 더 넣지 못하게 큐를 종료하고, 기존 항목을 처리한 뒤 join()을 기다린다. 그 다음 작업자 태스크의 종료를 확인한다. 큐의 종료와 태스크 그룹의 종료를 별도 단계로 다루면 어디서 데이터가 유실됐는지도 설명하기 쉬워진다.

QueueShutDown은 성공 영수증이 아니다. 생산자에게는 "더 받지 않는다"는 제어 신호이고, 소비자에게는 "이제 꺼낼 항목이 없다"는 제어 신호다. 실제 처리 성공은 각 업무의 결과와 task_done() 호출로 확인해야 한다.

그림 2. 두 모드 모두 새 put을 막지만, 기본 종료는 남은 항목을 처리하고 즉시 종료는 큐 안의 항목을 drain한다. 출처: docs.python.org/3/library/asyncio-queue.html#asyncio.Queue.shutdown.

4. 즉시 종료는 큐에 남은 항목을 버린다

queue.shutdown(immediate=True)는 장애 복구나 강제 중단에 가깝다. CPython 3.14.6 구현은 큐 안에 남아 있는 항목을 하나씩 꺼내 버리고, 버린 항목 수만큼 미완료 카운터를 줄인다. 그래서 세 항목이 모두 큐에 남아 있는 상태에서 즉시 종료하면 큐 길이와 미완료 카운터가 모두 0이 된다. 실제 업무 처리를 하나도 하지 않았어도 join()이 반환할 수 있다.

Python 문서가 즉시 종료와 join() 조합에 주의를 요구하는 이유다. 평소의 join()은 모든 항목에 task_done()이 대응됐다는 의미를 갖지만, 즉시 종료에서는 큐가 버린 항목도 카운터에서 빠진다. 따라서 await queue.join()이 끝났다는 한 줄만 보고 저장·전송·결제가 모두 처리됐다고 기록하면 안 된다.

강제 종료에서 버린 일을 다시 처리해야 한다면 별도 영속 원장이 필요하다. 큐에 넣기 전에 안정적인 작업 ID와 상태를 저장하고, 완료 시점을 멱등하게 갱신하는 방식이다. 메모리 큐의 join()은 프로세스 장애 뒤 복구 증거가 될 수 없다.

5. 실측해 보니 처리 중인 한 건은 그대로 남았다

여기서 한 가지 세부 동작이 눈에 띈다. 즉시 종료가 미완료 카운터 전체를 무조건 0으로 덮는 것은 아니다. CPython 구현은 "아직 큐 안에 남아 있어 drain된 항목"만 차감한다. 소비자가 이미 get()으로 가져간 항목은 큐 밖에 있으므로 그대로 미완료 상태다.

Python 3.14.6에서 작은 검사를 실행했다. a, b, c 세 항목을 넣고 a를 먼저 꺼낸 뒤 shutdown(immediate=True)를 호출했다. 큐에 남은 b, c는 버려져 qsize()가 0이 됐지만 미완료 카운터는 1로 남았다. 이 상태의 join()은 50밀리초 제한 안에 끝나지 않았다. 처리 중인 a에 대해 task_done()을 호출하자 그때 반환했다.

그림 3. 즉시 종료는 큐에 남은 두 항목만 차감했고, 이미 처리 중인 한 항목은 task_done 뒤에야 join을 풀었다. 출처: github.com/python/cpython/blob/v3.14.6/Lib/asyncio/queues.py#L253-L279.

이 검사는 CPython의 비공개 속성 _unfinished_tasks를 관찰용으로만 읽었다. 애플리케이션 코드가 이 속성에 의존해서는 안 된다. 다만 공개 API의 체감 동작은 분명하다. 즉시 종료는 대기열에 남은 일을 버릴 수 있지만, 이미 작업자에게 넘어간 일의 완료 책임까지 없애지는 않는다.

6. 운영 코드에서는 종료 이유를 상태로 남긴다

정상 배포 종료라면 shutdown()의 기본값을 쓰고 남은 작업을 처리한다. 장애가 번지는 상황이라 즉시 중단해야 한다면 immediate=True를 선택할 수 있지만, 이때는 join()을 완료 지표로 쓰지 않는다. 버린 항목 수, 처리 중이던 작업 ID, 재처리 여부를 따로 기록한다.

생산자와 소비자 태스크도 명시적으로 거둬야 한다. shutdown()은 큐에 막힌 put()과 get()을 깨우지만, 큐와 무관한 네트워크 호출에서 기다리는 태스크까지 취소하지 않는다. TaskGroup 같은 구조화된 동시성 도구로 태스크의 생성과 취소 범위를 묶고, 큐는 생산 중단과 drain 규칙만 맡기는 편이 낫다.

실전 점검 항목은 길지 않다.

• 새 생산을 막는 시점과 기존 항목을 버릴지 여부를 나눈다.
• get() 성공마다 정확히 한 번의 task_done()이 대응되게 한다.
• 즉시 종료에서는 join()을 업무 성공 증거로 사용하지 않는다.
• 외부 부작용이 있다면 작업 ID와 완료 상태를 큐 밖에 영속화한다.
• 큐 종료 뒤에도 살아 있는 생산자·소비자 태스크를 별도로 기다리거나 취소한다.

이번 검사는 처리량, 다수 작업자의 취소 경쟁, 다른 Python 구현체까지 측정한 벤치마크가 아니다. CPython 3.14.6의 문서·구현·테스트와 로컬 재현 결과를 맞춰 본 것이다. 결론도 그 범위로 좁혀야 한다. shutdown()은 센티널보다 선명한 종료 프로토콜을 제공하지만, 완료 증거와 복구 설계까지 대신하지는 않는다.

참고 자료

• Python 문서: asyncio queues
• CPython 3.14.6: Lib/asyncio/queues.py
• CPython 3.14.6: test_asyncio/test_queues.py
• CPython 이슈: Queue shutdown, gh-96471
• CPython PR: Add asyncio queue shutdown, gh-104228
• Python 문서: asyncio task groups and cancellation