FastAPIで非同期処理を理解する -FastAPIで安直にasyncしてはいけない-



## 自己紹介
初カキコども
**医師でエンジニア？もどきのikora**です

`python`については今まで趣味程度しか触ったことがなく、また`FastAPI`をこの一ヶ月前くらいから触り始めました。

エンジニアの方々にとっては当たり前の`非同期処理`というものについてあまり理解できてなかったので、FastAPIの挙動と共に理解しようということで忘備録的に書きました。

一応後半では、真面目に実験しているのでなにかの参考になるかなと思います。

素人に毛が生えたようなものなので、もし間違い等あったら優しめにご指摘いただくと嬉しいです☺️
始めての技術記事で拙いかもしれませんが、参考になりましたら幸いです。



## まず結論

FastAPIのマルチタスク性能・高速性は非同期処理によって実現されている

従ってAPI Endpoint配下の関数（パスオペレーション関数）の書き方で速さが変わり

**非同期処理 >>> 同期処理 >>>>>>>> 非同期定義 + 同期処理（async-lock）**

となる

## 前提

FastAPIは `Go` や`Node.js` に匹敵する速さと、開発速度・バグの少なさが特徴

またOpenAPI(Swagger UI)による自動ドキュメントで `スキーマ駆動開発` が高速に行える

FastAPIの速さの鍵は `Uvicorn` と `非同期処理(Starlette)` にある

## 非同期処理→並列と並行処理について

並行と並列処理は別物

ただし両方を組み合わせることで高速なレスポンスが行える

### 並列処理

FastAPIにおける並列処理は `Worker` と `Threads` で表現される

`Worker` は実質的にCPUのCore数だと思ってもらって良い（物理・仮想などもあるが）

```python
uvicorn main:app --workers 4 --threads 2
```

この場合、タスクは4 × 2 = 8列に並列して行うことになる

（この書き方はわかりやすく書いてるので実は `uvicorn run`では実行できない。
また後述する非同期処理が存在する場合はスレッド分けは非効率のため推奨されない）

### 並行処理
https://christina04.hatenablog.com/entry/fastapi-def-vs-async-def

FastAPIは内部で最適化されており、並行処理が施されている

並行処理とは`一つの Worker` がどれだけのタスクを同時に並行して行えるかという考え

といっても一人で同時に行ってるというのは `並列処理` や`スレッディング`と変わらないのでは？という疑問が発生するので簡単な設定で説明する

例えばデータベース（以下、DB）やストレージとの通信でレスポンスに１秒以上かかる処理があったとする
ちなみにこれらの待ち時間はたいてい通信時間やDB・ストレージの書き込み処理で、これらを総称して`I/Oバウンド（Input/Output）`と呼ぶ

APIサーバーとしてはレスポンスがいつ返ってくるかわからないので、１秒以上待って帰ってきた時にようやくクライアントに応答することができる。その際、いつレスポンスが返ってくるかわからないので `Worker` は**待機状態**になる

実際には `Worker` は複数人いるので１秒の待機でクライアントへのレスポンスがストップすることは無いが、上記の `Worker` と `Threads` では同時に８リクエストがあった時に、完全に一秒間ストップすることになるだろう・・

その問題を解決するのが `非同期処理` である 

`非同期処理` では何秒待つかわからない処理について `await` で知らせることで **「この処理は時間がかかりそうだから他の処理に移ろう」** ということを指示することができる

そのため、その待ち時間に他の処理に移ることができるのだ

また時間のかかる処理が終わったことをキャッチするために後述する `イベントループ` で監視（回遊）しているため、適切なタイミングでレスポンスを返すことができる


:::note
☝🏻日常的な例え話として、市役所に行った際、いつ呼ばれるかわからないのに窓口の前でずっと待機するようなこと（ `同期処理` ）はせず、番号が手渡され、時間がかかりそうなのでスマホで時間を潰そう→呼ばれたら窓口に向かう（ `非同期処理` ）という流れである
:::



## Pythonでの並行処理

ではPythonではどのように並行処理が施されているのか

簡単なコードを下記に示す

```python
async def my_task():
    # 非同期処理の内容を記述
    pass

task = loop.create_task(my_task()) #イベントループにtaskを登録

loop.run_until_complete(task) #イベントループを作成してRun
```

まず `async def` で関数定義する

この関数は `コルーチン関数` と呼ばれ、非同期処理を行うということを宣言している

```python
task = loop.create_task(my_task()) #イベントループにtaskを登録
loop.run_until_complete(task) #イベントループを作成してRun
```

この部分に関しては、先程のコルーチンで定義された `my_task関数` をイベントループに登録し、実行している部分である

この `コルーチン関数` で定義されイベントループに登録された関数については、呼び出す時に

```python
await my_task():
```

と書くことで **「時間のかかる処理だ」** と知らせることができ、一時停止することなく他のタスクに移ることができる

Pythonでは**イベントループに複数のコルーチン関数を突っ込む**ことで、処理が終了したか（レスポンスが帰ってきたか）を監視し、適切に `非同期処理` を実現している

## FastAPIと非同期処理

まずはFastAPIの一般的なエンドポイントのコードについて

```python
@app.get("/item/{item_id}")
async def get_item(item_id :int):
	item = db.get_item(item_id) #データベースからアイテムを取得し代入
	return item
```

エンドポイント直下の `get_item()` についてはFastAPIでは `パスオペレーション関数` と呼ばれている（API エンドポイントが呼ばれた際に実行される関数）

基本的にFastAPIのドキュメントでは `async def` と非同期処理を前提として書かれている

しかし非同期処理のページでは

https://fastapi.tiangolo.com/ja/async/

> データベース、API、ファイルシステムなどと通信し、`await` の使用をサポートしていないサードパーティライブラリ を使用している場合、次の様に、単に `def` を使用して通常通り *path operation 関数* を宣言してください:
> 
> 
> ```python
> @app.get('/')
> def results():
>     results = some_library()
>     return results
> ```
> 

とある通り、 `await` をサポートしていない場合は同期関数として定義する必要がある

なので先程のコードについては `await` が対応していれば

```python
@app.get("/item/{item_id}")
async def get_item(item_id :int):
	item = await db.get_item(item_id) #データベースからアイテムを取得し代入
	await my_task()
	return item
```

もし `await` が対応してなければ

```python
@app.get("/item/{item_id}")
def get_item(item_id :int):
	item = db.get_item(item_id) # データベースからアイテムを取得し代入
	my_task() # ちなみにもしmy_task()が非同期関数であればエラーが発生する
	return item
```

とする必要がある

ここで３つのパターンが考えられる

1. asyncで定義 + 中身も非同期関数(await)
2. defで定義 + 中身は同期関数(def)
3. asyncで定義 + 中身は同期関数(def)
4. ~~defで定義 + 中身は非同期関数~~

ちなみに４パターン目の ~~**defで定義 + 中身は非同期関数**~~

についてはそもそも同期関数内で非同期処理を定義できない→Pythonエラーになるので考えない

### 1. asyncで定義 + 中身も非同期関数(await)

コレは何も問題ない。FastAPIが一番推奨しているパターン

しかし非同期処理に対応していないライブラリがまだ存在するので、その場合は要検討

### 2. defで定義 + 中身は同期関数(def)

実はこの部分に関しては少し理解が面倒

FastAPIは並列・並行処理を適切に行うようになっており、もし`def`で定義した場合

`外部スレッドプール` というFastAPIの共同作業場にタスクが登録される

そのため**FastAPIが複数のWorkerを使ってよしなにタスクを並列・並行処理してくれる**

これは非同期処理よりも遅いが、しかし十分高速にさばくことができる

### 3. asyncで定義 + 中身は同期関数(def)

２と変わらないように見えて実は大きく異なる

今まで説明してきたように `FastAPIの非同期処理` については `単一のWorker` が如何にその処理を並行的に行うかということに焦点を当てている

そのため `async` でパスオペレーション関数を定義してしまうとその処理は `単一のWorker` が処理するという宣言をしてしまうのだ

２では作業場にたくさんのタスクが転がっており、複数のWorkerがよしなに処理してくれるのに対して、**３ではオーダーメイドのように一つのタスクに一つのWorkerが付きっきりになってしまう**のだ（そして待ち時間が発生した場合、当然そのWorkerは休憩することになる）

このため `I/Oバウンド` のような `ブロッキング処理` が発生すると単一のWorkerがそれにつきっきりになり後述の `Async-Lock` が発生してしまう

:::note
☝🏻 日常的な例え話として部屋の掃除を考えてみよう
１の非同期処理では、８人の作業者が各々休むことなくテキパキと動いている
２の同期処理では、リーダー役が８人の作業者に作業を与えてそこそこ動いている
３の場合は、８人の作業者が時々仕事を休みながら動いている
:::

## Async-Lock

https://zenn.dev/tk_resilie/articles/fastapi_async_sync

 `Async-Lock` : これは私が勝手に呼んでいる 

非同期処理のasyncでコルーチン関数を定義したにも関わらず、内部で同期処理が行われてる際に起こる**ブロッキング**

単一のWorkerで処理するため、待ち時間（`I/Oバウンド`）が発生した場合には完全にレスポンスが来るまで待機してしまう

**そのためリクエストが多重になった時に対応できない**

例えばWorkerが8人で１秒かかる処理があったとして、 `Async-Lock` の場合、秒間8個以上のリクエストが来るとパンクしてしまい、それ以上のリクエストについては待機列に並ぶしかなく、圧倒的に遅くなってしまう

## 実際の速度・多重度の測定

ということで

1. 非同期処理
2. 同期処理
3. Async-Lock

の３つについて実際に多重度を上げて速度を比較してみた

測定ツールはpythonで記述でき、WebUIでビジュアル化できる `locust` を使用した

https://locust.io

### 前提

- APIサーバーとDBサーバーはローカルに立てる
- APIはFastAPI, DBはelasticsearch
- この場合通信速度はほぼ0なので、sleep(0.05)などを挟むことで擬似的にI/Oバウンドを再現
- DBの操作・アクセスなどもI/Oバウンドに含まれるが、今回は小さいので無視する
- DBの負荷がMaxだとそちらがボトルネックになりAPI測定にならないので、DBには余裕をもたせるレベルのリクエスト数を送る
- 環境はMac Studio M1 Max(8 core)でworkersを8人にする
- GETはシンプルにIDを指定し一つのItemを取得する負荷
- GETは秒間約１回のリクエストで、一秒間に10人ずつ増やす設定、最大1000人（各ユーザーが秒間1リクエスト=秒間1000リクエスト）まで負荷をかける

### 1. 非同期処理

GET: asyncio.sleep(0.05) → I/Oバウンド（通信時間）が50msの設定
![async.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/584851/7c474254-0da6-b407-9b3c-142d7a05fc5a.png)



ユーザー数の増加とともにほぼ同数の処理をこなすことが出来ている

秒間1000リクエストでも余裕でさばけている
I/Oバウンド時間が50msだがレスポンスタイムもほぼそれに近いので、即時レスポンスが出来ている状態

### 2. 同期処理

GET: asyncio.sleep(0.05) → I/Oバウンド（通信時間）が50msの設定
![def.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/584851/f9969835-28cd-705f-7966-a9c64e9c1331.png)

FastAPIでさばけるリクエスト数が秒間700reqで頓挫してしまい、以降の増加については待ち時間になっている

非同期処理では I/Oバウンド時間(50ms)とほぼ同時に返せていたが、こちらは途中から待機列が出来ており、このレベルの負荷が続けば１秒近く待たされる可能性がある

### 3. Async-Lock
![async-lock.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/584851/05de8a1e-eeff-a7e9-3262-7ee7a39f6ece.png)



秒間15リクエストがやっとであり、その状態が続くと延々とレスポンスが伸び続け、ついにリクエストタイムエラー（赤の部分）が発生する

### 補足

ちなみにGETはかなり高速な処理に当たるのでI/Oバウンドはほぼ通信時間とみていいが、searchやPOST(create)などの処理になるとI/Oバウンド(待機時間)が加算されるのでもっと開きが出る

例えばI/Oバウンド（待機時間）が300msだった場合のGETについては
**非同期処理 > 同期処理 > Async-Lock**について

| 方法 | 処理速度(req/s) |
| --- | --- |
| 非同期処理 | 2000 |
| 同期処理 | 120  |
| Async-Lock | 2 |

という結果である(実際には非同期処理については2000reqを超えるとDBのほうが過負荷になりそれ以上測定できない)

またAsync-LockのPOST(create)では通信時間が0タイムであったとしても秒間40リクエスト（非同期通信では1500req/s可能）がやっとであり、 **通信時間が300msの設定ではほぼ秒間1リクエストしか出来ないという惨敗の結果**である

今回の実験では他のPOSTや様々な通信速度を総合すると

:::note
**非同期処理 > 同期処理 > Async-Lock**の処理速度については

`2000 > 100~500 > 1~15 (req/s)`

程度の開きが確認された
:::

またこれは**I/Oバウンドが発生すればするほど開きが大きくなる**

## 結論

今回は非常にシンプルな負荷だったが、現実世界とは違うことには注意しなければいけない


秒間1000リクエストはオーバーかもしれないが、現実世界では一番シンプルなID指定のGETのみが来るわけではない。POSTやsearchなど**ありとあらゆるI/Oバウンドが発生するリクエストが飛び交う**

DBの設定によっては通信時間を食う可能性もあり、また**画像のリクエスト・マネージドサービス・S3との通信はさらにボトルネックになるだろう**


一応補足すると、FastAPIのドキュメントがasync前提で書かれているため、同期処理については最適化の余地があるかもしれない
またローカルにAPIサーバー・DB両方立ち上げているため厳密な試験ではないことに注意してください

:::note warn
しかし上記結果からわかることは**Async-Lockについてはかなり早めに対策するべきである**ということ
:::

可能なら非同期処理に対応したライブラリの使用が望ましいが、難しい場合は同期処理で対応するのがbetterである