FastAPI 성능: 놓치고 있을 수 있는 숨겨진 스레드 풀 오버헤드

Source: Dev.to

문제 이해

FastAPI는 파이썬으로 고성능 API를 만들기에 뛰어난 프레임워크입니다. 비동기 기능, 자동 검증, 훌륭한 문서화 덕분에 사용하기 즐겁습니다. 하지만 미묘한 성능 문제가 종종 간과됩니다: 동기 의존성을 위한 불필요한 스레드‑풀 위임.

FastAPI가 의존성을 처리하는 방식

FastAPI는 async와 sync 호출 가능 객체를 구분합니다:

• async def 함수 – 이벤트 루프에서 바로 실행됩니다.
• def 함수 – anyio.to_thread.run_sync를 통해 스레드 풀로 전달됩니다.

이 동작은 경로‑연산 함수와 의존성 모두에 적용됩니다. 내부적으로 FastAPI는 다음과 같은 간단한 검사를 수행합니다:

import asyncio
from anyio import to_thread

# Simplified FastAPI logic
if asyncio.iscoroutinefunction(dependency):
    # Run directly in event loop
    result = await dependency()
else:
    # Send to thread pool
    result = await to_thread.run_sync(dependency)

클래스 생성자(__init__)는 항상 동기이기 때문에, 클래스‑기반 의존성은 항상 스레드 풀로 라우팅됩니다.

스레드 풀 오버헤드

• 기본 스레드 풀 크기: 40 스레드.
• 각 스레드‑풀 실행은 컨텍스트 스위칭, 스레드 동기화, 그리고 모든 스레드가 바쁠 때 발생할 수 있는 대기열을 초래합니다.

예시: 여러 클래스‑기반 의존성

from fastapi import Depends, FastAPI

app = FastAPI()

class QueryParams:
    def __init__(self, q: str | None = None, skip: int = 0, limit: int = 100):
        self.q = q
        self.skip = skip
        self.limit = limit

@app.get("/items/")
async def read_items(params: QueryParams = Depends()):
    return {"q": params.q, "skip": params.skip, "limit": params.limit}

각 요청은 단순히 값 할당만 하는 QueryParams 인스턴스를 스레드 풀에서 생성합니다.

엔드포인트에 이러한 의존성이 여러 개 있으면 오버헤드가 곱해집니다:

@app.get("/complex-endpoint/")
async def complex_operation(
    auth: AuthParams = Depends(),
    query: QueryParams = Depends(),
    pagination: PaginationParams = Depends(),
    filters: FilterParams = Depends(),
):
    pass  # Four dependencies → four thread‑pool tasks

동시 100개의 요청이 들어오면 400개의 스레드‑풀 작업이 대기하지만, 동시에 실행될 수 있는 것은 40개뿐이므로 지연이 급증합니다.

실제 영향

• 엔드포인트 50개
• 엔드포인트당 평균 3개의 클래스‑기반 의존성
• 초당 1 000 요청

→ 초당 약 150 000개의 불필요한 스레드‑풀 작업이 발생합니다. 각 작업이 빠르더라도 누적 오버헤드가 병목이 될 수 있습니다.

해결책: fastapi-async-safe-dependencies

스레드 풀을 우회하고 이벤트 루프에서 직접 실행하도록 특정 의존성을 표시하는 가벼운 라이브러리입니다.

설치

pip install fastapi-async-safe-dependencies

기본 사용법

from fastapi import Depends, FastAPI
from fastapi_async_safe import async_safe, init_app

app = FastAPI()
init_app(app)  # Initialize the library

@async_safe  # Mark as safe for async context
class QueryParams:
    def __init__(self, q: str | None = None, skip: int = 0, limit: int = 100):
        self.q = q
        self.skip = skip
        self.limit = limit

@app.get("/items/")
async def read_items(params: QueryParams = Depends()):
    return {"q": params.q, "skip": params.skip, "limit": params.limit}

무엇이 바뀌었나요?

1. 시작 시 init_app(app)을 호출합니다.
2. 의존성 클래스를 @async_safe로 데코레이트합니다.

내부 동작 원리

클래스에 @async_safe를 적용하면 라이브러리는 비동기 래퍼를 생성합니다:

# Simplified wrapper generated by @async_safe
async def _wrapper(**kwargs):
    return YourClass(**kwargs)  # Direct constructor call

래퍼가 코루틴이므로 asyncio.iscoroutinefunction이 True를 반환하고, FastAPI는 이를 이벤트 루프에서 바로 실행합니다—스레드 풀을 전혀 사용하지 않습니다.

init_app()은 모든 라우트와 의존성을 순회하면서 클래스 참조를 이러한 래퍼로 교체합니다. 래퍼 자체는 await를 수행하지 않으며, 동기 생성자를 즉시 실행합니다. 생성자가 블로킹되지 않을 때 안전합니다.

상속 지원

from fastapi_async_safe import async_safe

@async_safe
class BaseParams:
    def __init__(self, limit: int = 100):
        self.limit = min(limit, 1000)

class QueryParams(BaseParams):
    def __init__(self, q: str | None = None, **kwargs):
        super().__init__(**kwargs)
        self.q = q

서브클래스가 스레드 풀 실행이 필요할 경우(예: I/O 수행) @async_unsafe로 표시합니다:

from fastapi_async_safe import async_unsafe

@async_safe
class BaseParams:
    pass

@async_unsafe  # Will be sent to thread pool
class HeavyParams(BaseParams):
    def __init__(self):
        self.data = some_blocking_operation()

전역 옵트인

init_app(app, all_classes_safe=True)  # Treat all class‑based dependencies as async‑safe
# Use @async_unsafe only for exceptions

동기 함수와 함께 사용하기

데코레이터는 일반 함수에도 적용됩니다:

from fastapi_async_safe import async_safe

@async_safe
def get_common_params(q: str | None = None, skip: int = 0, limit: int = 100) -> dict:
    return {"q": q, "skip": skip, "limit": limit}

@app.get("/items/")
async def read_items(params: dict = Depends(get_common_params)):
    return params

벤치마크 및 결과

모범 사례

@async_safe를 사용해야 할 때

✅ 단순 데이터 클래스
✅ 파라미터 검증 클래스
✅ 설정 객체
✅ 블로킹되지 않는 유틸리티 함수
✅ Pydantic 모델 래퍼

❌ 사용 금지 대상:

• 데이터베이스 조회
• 파일 I/O
• 외부 API 호출
• CPU 집약적 계산
• 이벤트 루프를 실제로 블로킹하는 모든 작업

도입 전략

1. 작게 시작 – 가장 많이 호출되는 엔드포인트에 적용합니다.
2. 모니터링 – 지연이 개선되고 회귀가 없는지 확인합니다.
3. 점진적 확대 – 점차 더 많은 의존성을 async‑safe로 표시합니다.
4. 전역 옵트인 고려 – 충분히 자신이 있다면 all_classes_safe=True를 사용합니다.

테스트

기존 테스트는 그대로 동작합니다:

import pytest
from fastapi.testclient import TestClient

def test_endpoint():
    client = TestClient(app)
    response = client.get("/items/?q=test&limit=50")
    assert response.status_code == 200
    assert response.json()["q"] == "test"

주의 사항

• 조기 최적화 – 성능 문제가 확인될 때만 적용하세요.
• 블로킹 의존성 – 반드시 스레드 풀에 남겨두세요(@async_unsafe).
• 먼저 프로파일링 – uvicorn --log-level debug 같은 도구나 외부 프로파일러를 사용해 병목을 확인한 뒤 라이브러리를 적용하세요.