beautifulsoups.dev
Published on

파이썬 async 컨텍스트매니저로 리소스 누수 0

Authors

서버 사이드 파이썬에서 비동기 코드는 처리량을 올려주지만, 그만큼 리소스 관리가 까다로워집니다. 특히 async with로 감싸지 않은 네트워크 커넥션, 닫히지 않은 파일 디스크립터, 해제되지 않은 락이나 세마포어는 "가끔"이 아니라 "언젠가 반드시" 장애로 이어집니다. 장애 증상은 보통 Too many open files, 커넥션 슬롯 고갈, 워커가 영원히 대기하는 데드락 같은 형태로 나타납니다.

이 글에서는 파이썬의 비동기 컨텍스트 매니저를 두 가지 방식으로 정리합니다.

  • 클래스로 __aenter__와 __aexit__를 직접 구현하는 방법
  • contextlib.asynccontextmanager로 async generator 기반 컨텍스트 매니저를 만드는 방법

그리고 예외, 취소, 타임아웃 같은 현실적인 케이스에서 "정말로 누수가 0"이 되려면 무엇을 지켜야 하는지까지 다룹니다.

관련해서 운영 환경에서 리소스 고갈이 실제로 어떻게 드러나는지 감을 잡고 싶다면 리눅스 Too many open files 즉시 진단·해결도 함께 보면 좋습니다.

async 컨텍스트 매니저가 해결하는 핵심 문제

비동기 코드에서 누수는 보통 다음 패턴에서 발생합니다.

  • await 사이에 예외가 터져 close()나 release()가 실행되지 않음
  • asyncio.CancelledError로 태스크가 취소되며 정리 코드가 스킵됨
  • 타임아웃 처리 중 예외가 겹치며 정리 루틴이 중복 실행되거나 아예 실행되지 않음

async with는 진입과 종료를 언어 차원에서 보장합니다.

  • 진입 시 await obj.__aenter__()
  • 블록 종료 시 await obj.__aexit__(exc_type, exc, tb)

중요한 포인트는 "블록이 정상 종료하든 예외로 빠지든" 종료 훅이 호출된다는 점입니다. 즉, 정리 코드를 한 곳에 모아두면 누수 가능성을 급격히 낮출 수 있습니다.

방식 1: __aenter__·__aexit__를 가진 클래스로 만들기

클래스 기반 방식은 다음에 유리합니다.

  • 상태를 여러 필드로 들고 있어야 함
  • 재시도, 메트릭, 로깅, id 추적 등 부가 기능이 많음
  • 종료 로직을 여러 단계로 나눠야 함

아래 예시는 "동시성 제한 세마포어"와 "외부 리소스"를 함께 다루는 전형적인 패턴입니다.

import asyncio
from typing import Optional

class AsyncResource:
    async def open(self) -> None:
        await asyncio.sleep(0)  # 실제로는 소켓 연결, 파일 오픈 등

    async def close(self) -> None:
        await asyncio.sleep(0)

class ResourceLease:
    def __init__(self, sem: asyncio.Semaphore):
        self._sem = sem
        self._res: Optional[AsyncResource] = None

    async def __aenter__(self) -> AsyncResource:
        await self._sem.acquire()
        res = AsyncResource()
        try:
            await res.open()
        except Exception:
            # open 실패 시 세마포어를 반드시 반환
            self._sem.release()
            raise

        self._res = res
        return res

    async def __aexit__(self, exc_type, exc, tb) -> bool:
        # close는 실패할 수도 있으니, 세마포어 반환을 보장하기 위해 finally로 감싼다.
        try:
            if self._res is not None:
                await self._res.close()
        finally:
            self._res = None
            self._sem.release()

        # 예외를 삼키지 않고 그대로 전파
        return False


async def main() -> None:
    sem = asyncio.Semaphore(10)

    async with ResourceLease(sem) as res:
        await asyncio.sleep(0.1)

asyncio.run(main())

이 구현에서 누수를 막는 디테일

  • __aenter__에서 open()이 실패하면 release()를 즉시 호출하고 예외를 재전파합니다.
  • __aexit__는 close()가 실패하더라도 세마포어를 반환해야 하므로 finally로 감쌉니다.
  • __aexit__는 False를 반환해 예외를 숨기지 않습니다. 운영에서는 예외를 삼키는 컨텍스트 매니저가 문제를 더 키우는 경우가 많습니다.

취소와 타임아웃에서 안전하게: CancelledError를 특별 취급하지 말 것

비동기 서버에서는 취소가 흔합니다.

  • 클라이언트 연결 종료
  • 상위 타임아웃
  • 배포 시 graceful shutdown

이때 async with 블록 내부에서 태스크가 취소되면, 파이썬은 CancelledError를 발생시키며 블록을 빠져나갑니다. 중요한 점은 __aexit__가 호출된다는 것입니다.

다만 다음 실수는 흔합니다.

  • except Exception:만 잡아서 취소가 예상치 못하게 전파됨
  • 반대로 except BaseException:로 취소까지 삼켜서 태스크가 종료되지 않음

실무 권장 패턴은 이렇습니다.

  • 컨텍스트 매니저의 __aexit__에서는 취소를 "삼키지" 말고 정리만 하고 전파
  • 블록 내부에서 취소를 잡아야 한다면 CancelledError를 먼저 잡고 반드시 raise로 재전파
import asyncio

async def worker(sem: asyncio.Semaphore) -> None:
    try:
        async with ResourceLease(sem):
            await asyncio.sleep(10)
    except asyncio.CancelledError:
        # 정리는 __aexit__가 수행. 여기서는 로깅만 하고 반드시 재전파.
        raise

async def main() -> None:
    sem = asyncio.Semaphore(1)
    task = asyncio.create_task(worker(sem))
    await asyncio.sleep(0.1)
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        pass

asyncio.run(main())

방식 2: async generator 기반 asynccontextmanager

클래스 기반이 장황하게 느껴질 때가 많습니다. 특히 "리소스 획득 후 yield, 종료 시 정리" 구조는 generator가 표현력이 좋습니다.

contextlib.asynccontextmanager를 쓰면 yield 전이 __aenter__, yield 후가 __aexit__ 역할을 합니다.

import asyncio
from contextlib import asynccontextmanager

@asynccontextmanager
async def leased_resource(sem: asyncio.Semaphore):
    await sem.acquire()
    res = AsyncResource()
    try:
        await res.open()
        yield res
    finally:
        # finally는 정상 종료, 예외, 취소 모두에서 실행된다.
        try:
            await res.close()
        finally:
            sem.release()


async def main() -> None:
    sem = asyncio.Semaphore(10)

    async with leased_resource(sem) as res:
        await asyncio.sleep(0.1)

asyncio.run(main())

generator 방식의 핵심 규칙

  • yield는 정확히 한 번만 실행되어야 합니다.
  • 정리 로직은 반드시 finally에 둡니다.
  • yield 이전 단계에서 실패할 수 있는 작업이 있다면, 그 실패에서도 정리가 되도록 구조를 잡습니다.

이 패턴은 DB 커넥션, HTTP 클라이언트 세션, 임시 파일, 분산 락 등 대부분의 리소스에 적용할 수 있습니다.

__aenter__에서 무엇을 반환할지: 리소스 자체 vs 래퍼

async with X() as v:에서 v가 무엇인지가 API 품질을 좌우합니다.

  • 리소스 자체를 반환하면 사용자는 직관적으로 씁니다.
  • 래퍼를 반환하면 추가 기능을 제공할 수 있지만, 사용자가 내부 리소스 접근 방법을 배워야 합니다.

권장 기준은 다음과 같습니다.

  • 단일 리소스면 리소스 자체 반환
  • 여러 리소스를 묶거나, 호출 규약을 강제하고 싶으면 래퍼 반환

예를 들어 "트랜잭션"처럼 사용 규약이 중요한 경우 래퍼가 유리합니다. 이런 "규약 강제"는 스프링의 트랜잭션 전파 이슈와도 결이 비슷합니다. 관심 있다면 Spring Boot 3 @Transactional 전파·롤백 함정도 같이 읽어보면 좋습니다.

예외를 삼킬지 말지: __aexit__ 반환값의 의미

__aexit__가 True를 반환하면 예외가 억제됩니다. 즉, 블록 안에서 예외가 나도 바깥으로 전파되지 않습니다.

운영 코드에서는 대체로 False를 반환해 전파하는 편이 안전합니다.

  • 예외를 숨기면 장애 감지가 늦어짐
  • 상위 레벨 리트라이/서킷브레이커가 작동하지 않음

예외를 억제해야 하는 경우는 정말 제한적입니다. 예를 들어 "존재하지 않는 파일이면 그냥 넘어간다" 같은 정책성 처리 정도에만 사용하세요.

실전 패턴: HTTP 호출에 타임아웃과 취소를 섞어도 누수 없게

외부 API 호출은 타임아웃, 취소, 재시도가 섞이기 쉬워 리소스 누수의 온상입니다. 아래는 컨텍스트 매니저로 "세션 수명"을 강제하는 예시입니다. 라이브러리 의존을 피하기 위해 구조만 보여줍니다.

import asyncio
from contextlib import asynccontextmanager

class FakeSession:
    async def open(self) -> None:
        await asyncio.sleep(0)

    async def close(self) -> None:
        await asyncio.sleep(0)

    async def get(self, url: str) -> str:
        await asyncio.sleep(0.2)
        return "ok"

@asynccontextmanager
async def session_scope():
    s = FakeSession()
    await s.open()
    try:
        yield s
    finally:
        await s.close()

async def fetch(url: str) -> str:
    async with session_scope() as s:
        # 타임아웃이 나도 close는 보장된다.
        return await asyncio.wait_for(s.get(url), timeout=0.1)

async def main() -> None:
    try:
        await fetch("https://example.com")
    except asyncio.TimeoutError:
        pass

asyncio.run(main())

이 구조의 장점은 명확합니다.

  • 타임아웃이 어디서 나든 세션 종료는 반드시 실행
  • 호출자가 close()를 까먹을 여지가 없음
  • 세션을 공유하고 싶다면 session_scope를 상위로 올리는 것만으로 수명 제어 가능

외부 API에서 충돌이나 취소가 섞이는 케이스를 다루고 있다면, 오류가 났을 때 "정리"가 제대로 되는지부터 확인하는 것이 순서입니다. 관련 사례로 OpenAI Responses API 409 499 충돌 취소 오류 해결도 참고할 만합니다.

테스트로 누수 0을 검증하는 방법

"코드상으로 그럴듯"과 "운영에서 누수 0"은 다릅니다. 다음을 자동화하면 좋습니다.

  • 반복 실행 후 열린 파일 디스크립터 수가 증가하지 않는지 확인
  • 세마포어/락이 반환되지 않아 대기열이 쌓이지 않는지 확인
  • 취소를 강제로 걸어도 정리 로직이 실행되는지 확인

간단한 형태로는 아래처럼 취소를 대량으로 발생시키고, 세마포어 값이 복구되는지 체크할 수 있습니다.

import asyncio

async def job(sem: asyncio.Semaphore) -> None:
    async with leased_resource(sem):
        await asyncio.sleep(1)

async def main() -> None:
    sem = asyncio.Semaphore(5)
    tasks = [asyncio.create_task(job(sem)) for _ in range(50)]

    await asyncio.sleep(0.05)
    for t in tasks:
        t.cancel()

    for t in tasks:
        try:
            await t
        except asyncio.CancelledError:
            pass

    # 모든 작업이 취소되어도 세마포어는 원복되어야 한다.
    assert sem._value == 5

asyncio.run(main())

테스트에서 sem._value 같은 내부 필드에 접근하는 건 깔끔하진 않지만, 누수 검증 목적이라면 현실적인 선택입니다. 더 엄격히 하려면 계측 카운터를 별도로 두거나, 리소스 풀에서 현재 대여 수를 추적해도 됩니다.

체크리스트: async 컨텍스트 매니저로 누수 0 만들기

  • __aenter__에서 리소스 획득 후, 중간 실패 시 원복 로직을 반드시 넣는다
  • __aexit__ 또는 generator의 finally에서 정리 순서를 보장한다
  • 정리 중 예외가 나도 "다른 정리"가 스킵되지 않도록 try와 finally를 계층적으로 쓴다
  • 예외 억제는 최소화하고, 기본은 전파한다
  • 취소는 삼키지 말고 정리 후 재전파한다
  • 타임아웃, 취소, 예외를 섞은 스트레스 테스트로 누수를 검증한다

비동기 컨텍스트 매니저는 단순한 문법 설탕이 아니라, 운영 안정성을 올리는 구조적 장치입니다. 특히 파일 디스크립터나 커넥션 같은 "유한" 리소스는 누수가 누적되면 반드시 장애로 이어지므로, async with를 코딩 규약 수준으로 강제하는 것이 가장 효과적인 예방책입니다.

Discuss on Twitter • View on GitHub