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数だと思ってもらって良い（物理・仮想などもあるが）

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

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

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

並行処理

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

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

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

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

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

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

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

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

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

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

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

Pythonでの並行処理

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

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

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

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

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

まず async def で関数定義する

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

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

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

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

await my_task():

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

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

FastAPIと非同期処理

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

@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 と非同期処理を前提として書かれている

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

データベース、API、ファイルシステムなどと通信し、await の使用をサポートしていないサードパーティライブラリ を使用している場合、次の様に、単に def を使用して通常通り path operation 関数 を宣言してください:

@app.get('/')
def results():
    results = some_library()
    return results

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

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

@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 が対応してなければ

@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 が発生してしまう

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

Async-Lock

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

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

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

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

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

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

ということで

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

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

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

前提

• 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の設定

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

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

2. 同期処理

GET: asyncio.sleep(0.05) → I/Oバウンド（通信時間）が50msの設定

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

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

3. Async-Lock

秒間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や様々な通信速度を総合すると

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

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

程度の開きが確認された

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

結論

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

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

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

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

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

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