Python - Fastapi Long/slow Task timeout kill 해결 방안

FastAPI를 이용해서 Long/slow Task를 이용할 경우 많이 활용하는 방법은 바로 BackgroundTasks 기능을 활용하는 방법이다. 대부분의 경우 유용하게 add_task 만으로 별도의 작업을 관리할 수 있기 때문에 많이 유용하다.

from fastapi import BackgroundTasks

app = FastAPI()


@app.post("/long-task")

async def long_task(background_tasks: BackgroundTasks):
    # Long/Slow Task를 백그라운드에서 실행
    background_tasks.add_task(do_long_task)


async def do_long_task():
    # Long/Slow Task를 수행하는 함수
    ...

참고 문서 https://fastapi.tiangolo.com/tutorial/background-tasks/

하지만 이를 이용할 때 종종 Uvicorn/Gunicorn 에서 타임 아웃(Timeout) 관련 로그를 Woker에서 확인이 되는 경우, 이를 해소 할 수 있는 방법으로 유용한 수단 Gevent, run_in_threadpool, subprocess와 Celery를 활용하는 방법이라고 할 수 있다.

Celery는 Redis와 같은 방식으로 동작하기 때문에, 여기에서는 보다 가볍게 사용할 수 있는 Gevent, run_in_threadpool, subprocess에 대해서 정리해 보도록 하겠다.

run_in_threadpool 

run_in_threadpool 은 FastAPI에 존재하는 라이브러리에서 제공하는 기능으로 FastAPI에서 비동기적으로 작업을 실행하는 데 사용하는 함수입니다. 이 함수는 CPU 집약적인 작업이나 I/O 작업과 같은 비동기 작업을 처리하는 데 유용하다. 특히 메모리를 공유해야하는 상황에서는 더욱 유용하다고 할 수 있다.

Subprocess보다는 가볍게 동작할 수 있고, BackgroundTasks의 Worker 영향을 최소화 할 수 있다.

from fastapi.concurency import run_in_threadpool

app = FastAPI()

@app.post("/long-task")
async def long_task(request: Request, param1: int, param2: str):
    # 작업을 스레드 풀에 등록
    await run_in_threadpool(do_long_task, param1, param2)

    # 작업 결과를 반환
    return {"message": "Long task started"}

# 비동기적으로 실행할 작업
def do_long_task(param1, param2):
    ...

subprocess, gevent

subprocess는 별도의 프로세스를 생성하는 방식이기 때문에, 리소스라던가 프로세스를 생성하는 시간등의 손해가 있지만, Python의 GIL에서 자유롭고, 독립적인 실행이 가능하여 보다 큰 작업에 적합한 방식이라고 할 수 있다.

from fastapi import FastAPI, Request
from subprocess import Popen, PIPE

app = FastAPI()

@app.post("/long-task")
async def long_task(request: Request, command: str):

    # 작업을 백그라운드에서 실행, command = python3 test.py arg1 arg2
    do_long_task(command)

    # 204 No Content 응답 코드 반환
    return Response(status_code=204)

# 비동기적으로 실행할 작업
def do_long_task(command):
    Popen(command)

 

gevent는 subprocess를 함수 형태로 보다 사용하기 쉽게 만들어놓은 라이브러리이다. subprocess와 동일하게 GIL에서 자유롭다. gevent 는 아래와 같이 사용할 수 있다.

from fastapi import FastAPI, Request
from gevent import spawn

app = FastAPI()

@app.post("/long-task")
async def long_task(request: Request):


    # 작업을 Gevent에서 실행
    spawn(do_long_task)

    # 204 No Content 응답 코드 반환
    return Response(status_code=204)

# 비동기적으로 실행할 작업
def do_long_task():
    # ...