Blocking I/O in Async Handlers

API 서버에서 다른 서비스를 HTTP로 부르는 기능을 만들면서 requests를 다시 읽었다. 그리고 리뷰에서 “async def 안에서 requests를 부르면 이벤트 루프가 막힌다"는 지적을 받았다. 둘을 묶어 정리한다. 인용은 requests와 FastAPI 공식 문서 원문이다.

requests에서 실제로 쓰는 것

requests.get(url, headers=..., params=..., timeout=10)
requests.post(url, headers=..., json=..., timeout=10)
response.raise_for_status()
response.json()

params: 같은 키를 여러 번 보내기

You can also pass a list of items as a value

Note that any dictionary key whose value is None will not be added to the URL’s query string.

(Quickstart)

?state=queued&state=running처럼 같은 키를 반복해 받는 API가 있다. dict에 리스트를 넣어도 되고, 튜플 리스트를 넘겨도 된다. 튜플 리스트가 만들어질 URL과 모양이 더 가깝다.

from requests import Request

Request('GET', 'http://example.invalid/runs',
        params=[('state', 'queued'), ('state', 'running'), ('limit', '1')]).prepare().url
# 'http://example.invalid/runs?state=queued&state=running&limit=1'

Request('GET', 'http://example.invalid/runs',
        params={'state': ['queued', 'running'], 'x': None}).prepare().url
# 'http://example.invalid/runs?state=queued&state=running'

json: 본문과 Content-Type

If you need that header set and you don’t want to encode the dict yourself, you can also pass it directly using the json parameter

Note, the json parameter is ignored if either data or files is passed.

(Quickstart)

requests.post(url)과 requests.post(url, json={})은 다르다. 본문이 필수인 JSON API라면 앞의 것은 400이나 415를 받는다.

for kw in [{}, {'json': {}}, {'data': '{}'}]:
    p = Request('POST', 'http://example.invalid/x', **kw).prepare()
    print(kw, p.headers.get('Content-Type'), p.body)
# {} None None
# {'json': {}} application/json b'{}'
# {'data': '{}'} None {}

timeout: 없으면 무한정 기다린다

Nearly all production code should use this parameter in nearly all requests. Failure to do so can cause your program to hang indefinitely

If no timeout is specified explicitly, requests do not time out.

(Quickstart)

timeout은 응답 전체를 받는 데 걸리는 시간 상한이 아니다. 연결을 맺기까지 기다리는 시간(connect)과 바이트와 바이트 사이를 기다리는 시간(read)을 각각 제한한다. 숫자 하나를 넘기면 둘 다에 적용된다. 그래서 요청 전체에 실제로 걸린 시간은 timeout보다 길 수 있다.

requests.get(url, timeout=10)        # connect 10초, read 10초
requests.get(url, timeout=(3, 10))   # connect 3초, read 10초

raise_for_status()를 빼면 안 되는 이유

It should be noted that the success of the call to r.json() does not indicate the success of the response. Some servers may return a JSON object in a failed response

(Quickstart)

404와 함께 오류 JSON이 오면 r.json()은 그대로 성공한다. 그 뒤에 r.json()["items"]에서 엉뚱한 KeyError가 난다. 그러니 raise_for_status()를 먼저 부른다.

r.status_code  # 404
r.json()       # {'detail': 'not found'}
r.raise_for_status()  # requests.HTTPError

예외

ConnectionError(DNS 실패, 연결 거부), Timeout, HTTPError(raise_for_status())가 모두 requests.RequestException을 상속한다. 한 번에 잡으려면 RequestException을 잡는다. 테스트에서 side_effect로 실패를 흉내 낼 때도 이 이름들을 쓴다.

requests는 블로킹이다

With the default Transport Adapter in place, Requests does not provide any kind of non-blocking IO.

(Advanced Usage)

FastAPI 문서는 async def를 권하면서 조건을 붙인다.

it’s better to use async def unless your path operation functions use code that performs blocking I/O.

(FastAPI, Concurrency and async / await)

async def 안의 동기 호출 하나가 서버를 멈춘다

이벤트 루프는 await를 만날 때만 다른 일로 넘어간다. await가 없는 줄은 끝날 때까지 루프를 놓아 주지 않는다. async def 핸들러 안에서 requests.get(url, timeout=10)을 부르면, 상대가 느릴 때 최대 10초 동안 그 프로세스의 모든 요청이 기다린다. 다른 엔드포인트, 헬스 체크까지 포함해서다.

워커 프로세스마다 이벤트 루프가 하나씩 있으므로 영향 범위는 그 프로세스 전체다. 컨테이너 하나에 워커 하나만 띄웠다면 컨테이너 전체가 멈춘다.

그렇다고 타임아웃을 빼서 해결되지는 않는다. 기다리는 시간의 상한이 없어질 뿐이다.

직접 재 보기

동기 호출을 time.sleep(1)로 흉내 냈다.

import asyncio
import time

from fastapi import FastAPI

app = FastAPI()


def call_upstream() -> str:
    time.sleep(1)  # requests.get 같은 동기 호출
    return "ok"


@app.get("/blocking")
async def blocking():
    return {"result": call_upstream()}


@app.get("/to-thread")
async def to_thread():
    return {"result": await asyncio.to_thread(call_upstream)}


@app.get("/sync")
def sync():
    return {"result": call_upstream()}


@app.get("/health")
async def health():
    return {"ok": True}

uvicorn 워커 하나로 띄우고, 각 엔드포인트에 요청 5개를 동시에 보내면서 0.1초 뒤 /health를 한 번 불렀다.

/blocking  5 requests: 5.1s, /health: 4.92s
/to-thread 5 requests: 1.0s, /health: 0.00s
/sync      5 requests: 1.0s, /health: 0.00s

/blocking은 요청 5개가 하나씩 차례로 처리되어 5초가 걸렸고, 상관없는 /health도 5초 가까이 기다렸다.

고치는 방법 세 가지

1. 동기 호출을 스레드로 보낸다. 한 줄이면 된다.

   python copy

   result = await asyncio.to_thread(call_upstream)

2. 핸들러를 def로 선언한다. FastAPI는 def 핸들러를 스레드 풀에서 돌린다. 다만 핸들러가 이미 async 의존성(DB 세션 등)을 많이 쓰고 있다면 바꾸기 어렵다.

3. 비동기 클라이언트로 바꾼다. httpx.AsyncClient는 API가 requests와 비슷하다.

   python copy

   async with httpx.AsyncClient(timeout=10) as client:
       r = await client.get(url)
       r.raise_for_status()

전체를 한 방식으로 통일하는 것이 맞는 방향이더라도, 지금 고치는 변경에서 다룰 범위인지는 따로 판단한다. 당장은 1번으로 막고 나머지는 후속 리팩터링으로 남기는 선택도 있다.