FastAPI의 비동기 처리 방식

개요

Fastapi에서 async를 통해 비동기 함수를 어떻게 처리하는지 동작 방식에 대해서 학습하기 위해 작성한 글이며 코드를 통해 여러가지 상황에 대해서 실험을 진행한 내용에 대해서 설명한 글입니다.

FastAPI의 비동기 처리

 

FastAPI에서는 동기함수들을 비동기적으로 실행할 수 있게 해준다. 동기 함수를 이벤트 루프에 등록할시에는 이벤트 루프는 단일 스레드로 작동하기 때문에 동기 함수는 이벤트 루프의 메인 스레드를 차단하고, 다음 작업을 진행하지 못하는 상황이 생긴다. 따라서, 이벤트루프를 차단하지 않고 백그라운드에서 동기 함수를 실행할 수 있게 하기 위하여 분기처리를 진행합니다. 그래서 이벤트 루프는 다른 비동기 작업을 계속 진행할 수 있습니다.

• run_endpoint_function이라는 함수를 통해 코루틴 여부를 판단하고,
  • is_coroutine이 True일 경우, dependant.call이 비동기 함수라고 가정하고, 직접 비동기로 실행합니다 (await dependant.call(**values)).
  • is_coroutine이 False일 경우, dependant.call이 동기 함수라고 가정하고, 이를 비동기 이벤트 루프에서 실행하기 위해 run_in_threadpool을 사용합니다.
• is_coroutine = asyncio.iscoroutinefunction(dependant.call)
  • 해당 코드는 파이썬의 asyncio 라이브러리의 iscoroutinefunction 함수를 사용하여 dependant.call 함수가 코루틴 함수인지 여부를 검사하고, 그 결과를 is_coroutine 변수에 저장합니다.

본문

FastAPI는 비동기를 지원하여, 각 요청이 독립적으로 진행되고, 이는 효율적인 리소스 사용을 가능하게 합니다. 이걸 확인하기 위해 아래의 코드들을 구성해보았습니다. 각각의 코드는 비동기 함수에 동기적 time or 비동기적 time을 통해 구성하였다. 또한 API 요청을 동시에 보내기 위해 curl 를 활용합니다.

#test.txt
url ="http://127.0.0.1:8080/async"
url ="http://127.0.0.1:8080/async"

#command
curl --parallel --parallel-immediate --parallel-max 2 --config test.txt


"""
curl [options] [URL...]

--parallel: 이 옵션은 curl이 여러 요청을 동시에 병렬로 수행할 수 있게 합니다.

--parallel-immediate: 이 옵션을 사용하면 curl이 병렬 요청을 가능한 한 빨리 시작하도록 합니다.

--parallel-max 2: 이 옵션은 동시에 수행할 수 있는 최대 병렬 요청의 수를 제한합니다. 

--config test.txt: 이 옵션은 curl 명령어에 대한 추가 설정이나 다수의 URL, 헤더 등을 포함할 수 있는 구성 파일을 지정합니다. test.txt 파일 안에는 여러 curl 명령이나 설정이 포함될 수 있으며, 이 파일을 통해 입력을 자동화하고 복잡한 요청을 관리할 수 있습니다.

"""

 

실험을 통해 확인해 보니,

• 아래의 코드처럼 비동기 함수에 동기적인 sleep 함수를 사용하면, 이벤트 루프가 block되어 예상 동작 방식과는 다르게 동기적으로 처리가 진행됩니다.

from fastapi import FastAPI
import uvicorn
import time

app = FastAPI()

@app.get("/sync")
async def sync():
    print("시작")
    time.sleep(2)
    print("끝")

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8080)

"""
#예상 동작 방식
시작
시작
끝
끝
    
#실제 동작방식
시작
끝
INFO:     127.0.0.1:42626 - "GET /sync HTTP/1.1" 200 OK
시작
끝
INFO:     127.0.0.1:42642 - "GET /sync HTTP/1.1" 200 OK
"""

• 따라서 비동기의 이점을 활용하기 위해서는 아래의 코드처럼  await 를 사용한 비동기적인 asyncio.sleep() 함수를 통해 비동기적 이점을 활용해야합니다.

from fastapi import FastAPI
import uvicorn
import time
import asyncio

app = FastAPI()

@app.get("/async")
async def asyncs():
    print("시작")
    await asyncio.sleep(3)
    print("끝")
    
if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8080)

"""
시작
시작
끝
INFO:     127.0.0.1:40628 - "GET /async HTTP/1.1" 200 OK
끝
INFO:     127.0.0.1:40630 - "GET /async HTTP/1.1" 200 OK
"""

 

그렇다면 FastAPI 에서는 동기함수를 어떻게 처리할까?

• 동일한 방식으로 async가 아닌 def를 통해 동 time.sleep()을 사용하여 실험을 진행하였습니다.
• 엥? 예상한 동작방식이 아닌 비동기 함수에 await asyncio.sleep()처리한 것처럼 동작을 합니다.

from fastapi import FastAPI
import uvicorn
import time

app = FastAPI()

@app.get("/sync")
def sync():
    print("시작")
    time.sleep(2)
    print("끝")

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8080) 

"""
#예상 동작 방식
시작
끝
시작
끝

#실제 동작방식
시작
시작
끝
끝
INFO:     127.0.0.1:60594 - "GET /sync HTTP/1.1" 200 OK
INFO:     127.0.0.1:60596 - "GET /sync HTTP/1.1" 200 OK
"""

 

찾아보니 아래의 이슈문서에서 다음과 같은 내용을 발췌하였습니다.

• Use synchronous code (and sync path operation function), so FastAPI will run it in a seperate thread, and Python will yield control back when possible (e.g. when waiting for IO), or;
• Use asynchronous code (and async path operation function), so FastAPI will run it in the event loop, and your code will yield control back when possible.

https://github.com/tiangolo/fastapi/issues/5351#issuecomment-1238184673

Run async operations on separate threads · Issue #5351 · tiangolo/fastapi

 

• 위의 실험을 통해 비동기 함수는 단일 스레드를 통해 이벤트 루프에 동록되는것으로 확인 되었고, 동기 함수는 쓰레드풀을 통해 비동기형식으로 동작할 수 있다는 것을 확인하였습니다.
• 아래의 코드는 동기 함수와 비동기 함수의 쓰레드 ID 를 확인하는 과정입니다.

from fastapi import FastAPI
import time
import asyncio
import threading

app = FastAPI()

@app.get("/async")
async def asyncs():
    print("시작")
    print(threading.get_ident())
    await asyncio.sleep(3)
    print("끝")

@app.get("/sync")
def sync():
    print("시작")
    print(threading.get_ident())
    time.sleep(3)
    print("끝")

• 아래의 그림처럼 왼쪽은 비동기 함수의 결과값 및 쓰레드 ID 를 나타내며, 오른쪽은 동기 함수의 결과값 및 쓰레드 ID를 나타냅니다. 따라서 비동기는 단일 스레드 + 이벤트 루프를 통한 비동기 처리, 동기는 함수를 별도의 스레드에서 실행시키고, 그 결과를 비동기적으로 기다립니다. 이렇게 하면 비동기 이벤트 루프는 해당 동기 함수가 실행되는 동안에도 다른 비동기 작업을 계속 진행할 수 있습니다.

결과

결과적으로 동시성 코드는 별도의 쓰레드풀을 통해 작동하며 아래 발췌 내용을 보면  해당 쓰레드 풀은 총 40개로 구성된다고 합니다.

Now async stuff is handled by AnyIO. It has an internal dynamic thread pool. It creates threads up to a default max of 40, and it removes them when they are not used for a while.

 

https://github.com/tiangolo/fastapi/issues/2619

Behavior of synchronous path operation functions · Issue #2619 · tiangolo/fastapi