Table of contents

FastAPI에서 asyncio.create_task() vs ThreadPoolExecutor

FastAPI에서 SSE나 httpx.AsyncClient() 기반 스트리밍을 유지하면서 백그라운드 작업을 수행해야 하는 경우가 생겼다. 현재 구조는 다음과 같다.

asyncio의 기본 구조

FastAPI의 async endpoint는 보통 하나의 event loop 위에서 동작한다. 구조를 단순화하면 다음과 같다.

FastAPI Process
 └─ Event Loop
     ├─ request coroutine
     ├─ SSE coroutine
     ├─ websocket coroutine
     └─ background coroutine

핵심: 모든 async task는 같은 event loop를 공유한다. 따라서 loop 안에서 blocking이 발생하면 SSE·다른 API 요청·스트리밍까지 함께 지연될 수 있다.

이 저장소에서는 두 경로를 분리한다.

asyncio.create_task()란?

asyncio.create_task()는 새 OS thread나 process를 만드는 것이 아니다.
현재 event loop에 coroutine을 등록해 동시에 진행하게 하는 것이다.

import asyncio

async def background():
    await asyncio.sleep(5)
    print("done")

async def main():
    asyncio.create_task(background())
    return "ok"

이 경우:

• 새로운 OS thread 생성 없음
• 새로운 process 생성 없음
• 같은 event loop에서 스케줄링됨

pure async I/O(await 기반)만 있다면 가장 가볍고 적합한 방식이다.

create_task의 문제점

task 내부에 blocking 코드가 들어가면 event loop 전체가 멈출 수 있다.

async def background():
    import time
    time.sleep(10)  # blocking — loop 정지

async def background():
    df = pandas_heavy_work()  # GIL + CPU — loop 지연

발생 가능한 영향:

• SSE heartbeat·chunk 전송 지연
• 다른 요청 latency 증가
• websocket timeout
• upstream SSE 프록시 지연

즉, 같은 loop 안에서 blocking이 실행되면 async 시스템 전체에 영향이 간다.

주의: google-cloud-bigquery, google-cloud-storage 등 공식 클라이언트는 대부분 동기(sync) API다. async def 안에서 그대로 호출하면 create_task로 감싸도 loop를 막는다.

ThreadPoolExecutor란?

ThreadPoolExecutor는 blocking 작업을 별도 thread에서 실행한다.

Event Loop
 ├─ SSE (httpx async stream)
 ├─ API request
 └─ ThreadPoolExecutor
     ├─ thread 1  ← BQ job
     ├─ thread 2  ← GCS upload
     └─ thread 3  ← pandas transform

• event loop는 계속 살아 있음
• blocking 작업만 thread로 분리
• loop.run_in_executor(executor, fn, *args)로 결과를 기다리거나 fire-and-forget 가능

현재 workload 특성

전체적으로 CPU-heavy보다 I/O mixed workload에 가깝다.
목표는 SSE·async 스트리밍 안정성 — event loop를 blocking하지 않는 것이다.

왜 ProcessPoolExecutor 대신 ThreadPoolExecutor인가

1. BQ/GCS는 대부분 I/O 대기

BigQuery·GCS 호출 시간의 상당 부분은 네트워크 대기다. CPU를 계속 점유하는 작업이 아니므로 thread 기반 처리에 적합하다.

2. pandas 작업이 무겁지 않음

현재 pandas 사용은 단순 transform·소규모 DataFrame 수준이라 process 분리까지는 필요하지 않다.

3. SSE 안정성 확보가 목적

ThreadPoolExecutor만으로도 SSE, websocket, httpx async stream을 event loop에서 분리해 보호할 수 있다.

4. ProcessPoolExecutor 대비 구현 단순

process 기반은 다음 부담이 있다.

• pickle·직렬화 필요
• 메모리·startup 비용 증가
• 공유 상태(예: in-memory job status) 접근 복잡

thread는 메모리 공유·기존 객체 재사용이 쉽고 구현이 단순하다.

선택적 process pool

pandas가 커지거나 CPU-bound가 지배적이면 ProcessPoolExecutor를 고려할 수 있다.
이 저장소는 환경 변수 USE_PROCESS_POOL=true로 전환 가능하다 (app/core/executor.py).

FastAPI 적용 예시

1. Executor 생성 (앱 lifespan)

# app/core/executor.py
from concurrent.futures import Executor, ProcessPoolExecutor, ThreadPoolExecutor

def create_executor(settings: Settings) -> Executor:
    if settings.use_process_pool:
        return ProcessPoolExecutor(max_workers=settings.background_workers)
    return ThreadPoolExecutor(
        max_workers=settings.background_workers,
        thread_name_prefix="bg-job",
    )

# app/main.py
@asynccontextmanager
async def lifespan(app: FastAPI):
    settings = get_settings()
    app.state.executor = create_executor(settings)
    yield
    app.state.executor.shutdown(wait=True, cancel_futures=False)

앱 종료 시 pool을 정리해 thread/process leak을 방지한다.

2. blocking 작업 정의 (sync 함수)

thread/process pool 안에서는 sync 함수로 작성한다. await가 필요 없다.

# app/services/background_job.py
def run_background_job(job_id: str, params: dict[str, Any]) -> None:
    # sync BigQuery / GCS / pandas
    ...

3. endpoint에서 실행

# app/routers/jobs.py
@router.post("/jobs", response_model=JobResponse)
async def create_job(req: JobRequest, request: Request) -> JobResponse:
    job_id = str(uuid.uuid4())
    loop = asyncio.get_running_loop()
    executor = request.app.state.executor
    loop.run_in_executor(executor, run_background_job, job_id, req.model_dump())
    return JobResponse(status="ok", job_id=job_id)

동작 요약:

• API는 즉시 {"status": "ok", "job_id": "..."} 반환
• 실제 BQ/GCS/pandas는 pool thread(또는 process)에서 실행
• event loop는 SSE·다른 async 요청 처리 가능

4. SSE는 pure async 유지

# app/services/sse_proxy.py
async with httpx.AsyncClient(timeout=timeout) as client:
    async with client.stream(...) as response:
        async for chunk in response.aiter_bytes():
            yield chunk

SSE 경로에는 sync blocking 호출을 넣지 않는다.

환경 변수

BACKGROUND_WORKERS=4
USE_PROCESS_POOL=false

worker란 무엇인가

worker는 문맥에 따라 의미가 다르다. 혼동하지 않도록 구분한다.

1. uvicorn worker

uvicorn app.main:app --workers 2

여기서 worker = FastAPI 서버 프로세스 개수.

Process 1  
  └─ event loop  (독립)
Process 2  
  └─ event loop  (독립)

• CPU 멀티코어 활용·프로세스 격리에 유리
• in-memory job 상태·SSE 연결은 프로세스 간 공유되지 않음
• long-lived SSE는 sticky session 없이 --workers 2 이상이면 주의

2. ThreadPoolExecutor worker

ThreadPoolExecutor(max_workers=4)

여기서 worker = 동시에 실행 가능한 background 작업 수(thread 개수).

• 한 프로세스·한 event loop 안에서의 백그라운드 동시성 상한
• uvicorn worker 수와는 별개로 튜닝한다

max_workers 산정

workload가 I/O mixed이면 CPU 코어 수에 엄격히 맞출 필요는 없다.

이 프로젝트 기준

기본값 BACKGROUND_WORKERS=4 (app/config.py).

이유:

• SSE 안정성 확보에 충분한 여유
• BQ/GCS 대기 시간 동안 다른 job 실행 가능
• thread 과다·외부 API 동시 호출 폭증 위험 완화
• 구현·운영 단순

너무 크게 잡으면 안 되는 이유

• thread마다 메모리·스택 비용
• context switching 증가
• GIL 경쟁 (CPU-bound 구간)
• BigQuery/GCS quota·rate limit 압박

작게 시작 → latency·queue·에러율 모니터링 → 필요 시 증가가 안전하다.

다른 선택지와 비교

이 저장소는 단일 프로세스 + SSE 동시 유지가 목표이므로, 외부 큐 없이 ThreadPoolExecutor + run_in_executor가 현실적인 균형점이다.

최종 정리

asyncio.create_task()

ThreadPoolExecutor

결론

현재 workload는 BQ/GCS 중심이고 pandas는 무겁지 않으며, SSE 안정성이 중요하다.
따라서 blocking 작업은 ThreadPoolExecutor + run_in_executor로 event loop 밖 thread에 분리하는 방식이 가장 현실적이고 단순한 선택이다.

SSE·upstream 프록시는 async만 사용하고, 무거운 GCP/pandas 작업은 **POST /jobs**처럼 pool로 격리하는 이중 경로를 유지하는 것이 권장 패턴이다.