Pythonの処理をThreadに渡してもプログラム全体の動きが止まる

Question

実現したいこと

Pythonで文字列を暗号化、復号化するプログラムを作成しました
そのEncrypt, Decrypt処理では特に外部との通信などは行っておらず完全にローカルのみで解決するコードであり、その処理をFastAPIでEncryptとDecryptをapiで処理できるようにしたいと思い、単純にpostリクエストで渡されたbodyの値を暗号化する関数に渡しています。
そこで、一件ずつ暗号化や復号化が行われてしまい非同期的に処理が行われなかったためasyncio.to_threadを使用してみても特に早くならないどころか逆に遅くなっているように感じました
そこで、これを効率的に捌く方法を教えてください

前提

こういった場合はおそらくCPUバウンドと呼ばれる処理なのでしょうか？
こういった場合にはどういった対処が好ましいのでしょうか？Processに渡すべきなのかPython内で効率的に処理を捌く方法があるのか、それとももうGoなどに書きなおすべきなのか
個人的にパパっと書き直せる処理ではないので書き直しや大幅なコード改善は暗号化が正常に動作しなくなってしまう場合や別サービスとの連携もあり、この形式でなくてはならないという決まりがあるため書き直すなどは不可能です。
型付け程度であれば可能です。

またPythonのバージョンを3.11にする、や別言語への簡単な書き換え方法などがあればそういった方法でも問題ありません
実行環境の変更などは問題ありません

またencrypt.pyに関してはgzip compressやforループAES暗号化を行っているだけです。
まずencrypt処理でそれだけ時間がかかるのはおかしいなどは敢えて複雑な処理で複雑な暗号化を行っていますのでご理解ください
CPUバウンドを効率的に捌く方法を教えていただければと思います。

発生している問題・エラーメッセージ

エラー無し、一件ずつ処理されるため効率化の方法

該当のソースコード

from fastapi import FastAPI
from pydantic import BaseModel
from utils.encrypt import Encrypt
import asyncio

app = FastAPI()

@app.get("/")
async def root():
    return {"message": "Hello World"}

class EncryptRequest(BaseModel):
    body: str

@app.post("/encrypt/encrypt")
async def encrypt_encrypt(request: EncryptRequest):
    encrypted = await asyncio.to_thread(Encrypt().encrypt, request.body) #CPUバウンド 感覚で0.5秒ほど
    return {"encrypted": encrypted}

@app.post("/encrypt/decrypt")
async def encrypt_decrypt(request: EncryptRequest):
    return {"decrypted": Encrypt().decrypt(request.body)}

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

試したこと

asyncio.to_threadに渡す、これは効果的ではありませんでした
同様にloop.run_in_executorやconcurrent.futures.ThreadPoolExecutorやthreadingについても効果が無い(むしろThreadでは逆効果)と言えるでしょう

loop.run_in_executorにconcurrent.futures.ProcessPoolExecutorを渡し、負荷テストを行った場合Threadに処理を渡すよりはResponseを早く得ることが出来ましたが処理内容に対して1processを発行してしまっては処理内容に対してコストが高すぎるためマシ程度であり求めるほどの効果は得られませんでした

補足情報（FW/ツールのバージョンなど）

Python 3.9.13

Comments

[dameo]

足りないコードだけ提示されて、速度が～って言われてもねｗ

[KohnoseLami]

足りないコードだけ提示とはどういう意味でしょうか？
全てを書けということでしょうか？

Teratail上で共有できるほど短いコードではなく書いてある通りEncryptを行う処理に関してはほぼ変更不可能であり、書いても書かなくても一切問題が無いと判断したため書いていません
Encryptの処理を大幅に書き換えて最適化することは出来ません
それを書いたうえで型を付ける程度の簡単な変更は可能と書いてあります。

全てのコードをTeratail上で乗せるにはセキュリティ上の問題やそもそもとんでもない量でありページが長くなってしまうため省略しています。

もし自分でテストが出来ないから等の理由でコードが必要なのであればCPUバウンドをあえて発生されるダミーコードをこちらで用意して書きますのでそういった内容のコメントを書いていただければと思います。

自分で調べた限りではCython、つまりPythonの記法を使用してCに書き換える、やNumbaを使用する、に関しては大幅な書き換えであり引数にUnion型などの動的な型引数を渡したりするため不可能となっております。
少なからずこちらのCython, Numbaなどの実装に関しては「型付け程度であれば可能」という前提の項目に該当していないためこちらも省略させていただきました

つまりPython内で高速化するJITコンパイル、CythonでのC実装、他言語への手動書き換え、Multiprocessingでのプロセス分散、基本的に調べて出てくるような高速化が全て出来ない場合の手段を何か知っている方が居ればそれを回答した頂ければという思いで投稿させていただきました

そのため私が求めている回答はコードをほぼ書き換えずにPython 3.11などの高速化プロジェクトが行われ始めたバージョンに移動する、やmultiprocessingではなく他の何か処理を分散させられるモジュールを知っているのならばそれを使用する方法などFastAPIの呼び出している側でどうにか高速化を行いたいというのが目的です。
実行環境を変更するよりもEncryptファイルを書き換える方が大変なためPython 3.12等のテスト段階である実行環境で実行することも全く問題ございません
また、無理なのでしたら無理という回答をしていただければと思います。
正直無理難題であるのは承知です、私が無理だろうなと判断して人に頼るしかないと思ったので質問を建てました

コードの問題の箇所でしっかりとコメントアウトを入れてCPUバウンドが起きている場所とどれくらいの時間か書いていますのでそちらを参考までに
その処理をダミーな適当なCPUバウンドな処理に書き換えれば良いと思います。

これらを何かご存じであれば経験などから教えていただければとてもうれしく思います。
最後にそれでも本当のコードが必要なのでしたら何故必要なのかを詳しく教えていただければ幸いです、私にはまったく理由が分かりません

[dameo]

長文過ぎて読んでませんが、何をしたら載せられないほどになるのか想像できません。
linux環境想定でbashスクリプトで環境構築し動くサンプルコードを書いてみました。

python3 -m venv env
. env/bin/activate
pip install -U pip
pip install -U setuptools
pip install cryptography==40.0.2 fastapi==0.95.1 uvicorn==0.21.1
cat >hoge.py <<EOF
from fastapi import FastAPI, Request
from pydantic import BaseModel
from cryptography.fernet import Fernet
import time

key = Fernet.generate_key()
f = Fernet(key)
app = FastAPI()

def measure(func, args):
s = time.perf_counter()
ret = func(*args)
e = time.perf_counter()
return (ret, e - s)

@app.get("/")
async def root():
return {"message": "Hello World"}

class EncryptRequest(BaseModel):
body: str

@app.post("/encrypt/encrypt")
async def encrypt_encrypt(request: EncryptRequest):
ret, t = measure(f.encrypt, [request.body.encode('utf8')])
return {"encrypted": ret, "time": t}

@app.post("/encrypt/decrypt")
async def encrypt_decrypt(request: EncryptRequest):
ret, t = measure(f.decrypt, [request.body.encode('utf8')])
return {"decrypted": ret, "time": t}

if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
EOF
python hoge.py&
sleep 5
for _ in `seq 10`; do
wget -q -O - --method=post --header 'Content-Type: application/json; charset=utf-8' --body-data='{"body":"こんにちは、世界！"}' 'http://localhost:8000/encrypt/encrypt'
done
kill $(jobs -p)
wait
deactivate

[KohnoseLami]

しっかりと質問内容やコメントを読んでいないならもうコメントや回答なさらないでください
的外れな内容を送られても困りますので

暗号化に関してはTikTokなどでも使用されているアルゴリズムでありとても複雑なものとなっております
それが原因となり時間もかかる処理になっています
暗号化の部分に関しては別の暗号化方法などは求めておらず既に既存の暗号化方法を効率的に捌く処理方法を聞いています

その仮に書かれたコードのmeaure関数がCPUバウンドな処理で0.5秒ほどかかると想定してどのように効率的に捌けるかを回答として求めています

[dameo]

普通に書けますよね？？？サンプルくらいｗ

[KohnoseLami]

はい、サンプルくらいかけると思ったので質問本文では前述の理由により省略しております
それを最初から汲み取って貰えなかったようなので説明して自分でそのサンプルを使って検証してねということですね
そのご自身で書かれたコードの暗号化関数の呼び出しをさらに効率的に捌く方法を質問本文またはコメントに記載した方法を除き別の効率的な捌き方をご存知であるならば解答欄にお願いします

[dameo]

大丈夫ですか？あなたの問題なんですよ？サンプル書けるなら書いてくださいｗ
速度がどうのってサンプルもなしに言われても困るでしょｗ
私が求めているのは回答に必要な質問の修正ですｗ
検証すべき問題があなたが書いた質問文章の中にあると言ってるのですよｗ

[KohnoseLami]

では最初からどうぞそうコメントなさってください
伝えたいことのみを書けば良いのにずっと煽り口調で、伝わりませんよ
最初からそう言われていれば追加しましたよ
私は元から言語の仕様上無理だろうと思って誰か同様の経験があれば経験談から回答を貰えたらラッキー程度で話していたのでそもそもコードを書く必要性すら感じていませんでした
質問としては「CPUバウンドを効率的に捌く方法を教えて」だけで問題ないと感じています

あなたがこの質問にサンプルを必要と感じたのなら1回目のコメントの返答の時点でそう仰ってくださいね
私はそう聞いたのですから

[dameo]

最初のコメントに「足りないコードだけ提示されて、速度が～って言われてもねｗ」って書いてますよｗ
他人と共有して何かを解決するつもりならすぐに動くコードが必要ですｗ
特に速度要因なら再現できないと話になりませんｗ
早い話があなたの検証能力を疑っているわけですｗ
原因がそこなら回答しても何の意味もないのでｗ

[KohnoseLami]

最初からというのは少し間違ってましたね
言い換えると
それを言われた後にに何故書くことが出来ないのかと何故必要なのか？を聞き、検証が必要ということならば代わりのサンプルを用意した方が良いですか？
時いた訳です
その次にこのコメントが来てれば済んでいた話なんですがどうやらコメントを全て読まなかったようで噛み合わないコメントが来ましたので、、、

先程も言った通りTikTokなどでも使用されているアルゴリズムでセキュリティや処理工程の多さから実務で使用しているコードは提示できません
CPUバウンドが原因であることは確定ですがそれを信頼できないのであれば先程から言ってる通り別に回答やコメントもなさらなくて良いので

別にこの場はあなただけの場所では無いので
こういうこと言うと回答つかなくなると思いますが代わりは結局いくらでもいるような場所なので別に最初の時点でこちらが提示できないと言った時点であなたが回答する価値を見いだせないのであれば別に回答もコメントもしなくて良いのではと私は思います

このやり取りが楽しいなら別ですが煽り口調からするに不快なら無視すれば良いと思います

[dameo]

不快ではないですよｗ 業務なら他人の力を借りてはダメですねｗ プロジェクトメンバ以外に許可なく相談するなら、普通に守秘義務違反ですｗ
「TikTokなどでも使用されているアルゴリズムでセキュリティや処理工程の多さから実務で使用しているコード」(嘘くさいけど)を求めてはいませんが、CPUバウンドが原因であることは確定できるサンプルを提示し、そこでスレッド/プロセスプールを使って検証したコードと、結果も全て提示しないと、信用できないと言ってるだけですよｗ

そして、それらを提示できるなら、恐らくこの質問は発生しません。ようはあなたが検証したくないがために、他人に検証を委ねているように見えると言ってるわけです。もしそうなら、他人を利用したいだけのただのワガママですよね？ｗ

[KohnoseLami]

それならば良かったです
1人のプロジェクトなのでそこら辺はご心配なく
使えるものは使って近道した方が楽なので

TikTokでも使用されているというのに関しましては別に真偽は私もどちらでも良いのですがGithubなどでTTEncryptとお調べ下さい
こちらのTTEncryptの処理をほぼ流用という形で採用させてもらっているのでコード自体にかなりの価値や公開してしまうと悪用される恐れもあるため公開ができないというのが詳細な状態です

「そこでスレッド/プロセスプールを使って検証したコードと、結果も全て提示しないと、信用できないと言ってるだけですよｗ」に関してですが初めて言われましたね

本物のEncrypt処理はコード提示ができないためCPUバウンドを発生させるダミーで検証した結果とコードを書けば良いのですか？
それでは結局本来のコードとは処理が異なるので意味がある検証と言えるのでしょうか？
少なからずそもそもの話私はコードを効率化させたいのではなくCPUバウンドを効率良く処理する方法を聞いているわけなので別にコードなんてどうでも良いと思うのですが、、、

何故そう思うのでしょうか？
私は事実としてThreadでもProcessでも期待に沿うような結果は出ていません
検証したからこそasyncio to_threadやrun_in_executorなどを書いているわけです
もしかしたら効率良くの感覚に相違があるのかも知れません
私はAPIに対して1000req/1sec程度で負荷をかけています
実際の想定ではこれ以上のリクエストが飛んでくる予定なので余裕で裁けなくちゃいけないわけですがProcessでは1000Processなんか起動したらCPU使用率がパンクしてしまうのでProcessを使用できないわけです

[dameo]

仕事ではなくないですか？ｗ
正式に公開されている実装コードも存在してないですよね？ｗ
非公開なコードを正式にもらえているならそもそも話相手いるので聞けばいいですｗ
リアルタイムで非公開コードを悪用しようとしているか、そもそも大嘘であることになりますねｗ

プールを使った検証については元のサンプルが動かないのでそこまで話がいってないだけｗ
公開出来ないコードについて聞くのであれば元のコードの特性をそのままにしたダミーコードが必要です(当たり前)ｗ
検証してればよほど能力不足でなければ分かるはずだからですよｗ

[KohnoseLami]

仕事では無いですね
適当に書きましたw
はい、もちろん公開されていません
アプリ内に含まれている非公開コードです
そこを突かれたら特にもう言うこともなかったので言わずに書いてます
悪用してるかしてないかはここでいくら悪用していないとは言ってもどうとでもなる感あるのでとりあえずしていないということを言葉だけですが示しておきます
趣味程度で分解して拝借しただけで悪用はしていません
これでも悪用していると少しでも考えるならこれ以上アドバイスをされると悪用を協力しているということになりますので

では、あなたに証明することは不可能ですね
私の検証上Encrypt側を変更しない限りこれ以上の改善は見込めない、スレッドもプロセスも逆効果もしくは効率的とは言いきれない程度の改善だったので間違いなくあまり好ましい結果とは言えませんでした

ちなみに何度も言うようで申し訳ないのですがCPUバウンドを効率的に行う方法を聞いていますのでそれで納得できないのであればそこまでですね
私はasyncio.to_threadやrun_in_executorを用いてしっかりと検証を行っていますのでコードの問題ではなく言語上の限界だと私は考えます

一応では検証結果の秒数をせておきます
1000回計算時の結果で
syncな場合 81秒
threadな場合 91秒
processな場合 19秒

だいたいサイトの情報通りCPUバウンドの処理を行った場合と言えるでしょう
ですがこの場合1000reqが一気に来た場合processでも単純に計算すれば19秒もresponseを返すのにかかってしまう上わざわざProcessを発行してるため余分なCPUパワーを使いすぎてあまりにも実用に耐えず、また想定よりも効率的とは言えません
以上のことから原因はEncrypt処理でありCPUバウンドが原因、そしてProcessでも期待する結果が得られないということになります
概ねGoogleなどでヒットする結果と同等の速度と言えるでしょう
そのためこの方法以外での高速化を聞いています
まず1個生成するのに正確に測ったところ0.08秒かかっておりそもそもこの時点で繰り返しの計算や値の取り出しなどしか行っていないにもかかわらず時間がかかりすぎていると感じます
このことからPythonの限界なのだと思い無理難題としてこの質問を建てました
NumbaやCythonがこの状況ではいちばん好ましいのでしょう
ですがそれはしたくないというここはワガママもしくはラッキーが訪れて欲しいという願いですね

[dameo]

CPU依存な処理である以上、コア数以上の並列化は遅くなる方向の効果しかありません。
プロセスにしたときにちゃんと高速化が見られてるということで、質問の意味がなかったことが分かりましたね。
cpythonはnativeを混ぜない限り、Threadで並列化してもCPUがボトルネックになるような処理は速くなりません。
その後の解釈は間違っていて、通常プールの最大値は決めるもので、ある程度以上の生成コストはゼロで、プール以上のリクエストは単純に待たされます。

無駄に問答をするよりは、素直に質問し、詳細に書きましょう。あとダミーくらいすぐ書けますよ？
他の人のために書いてください。
最後に補足。fastapiは、asyncにすると非同期になりますが、asyncを外すとスレッドプールを使います。

なお限られたリソースで高速化を目指すのであれば、pythonを使わない選択肢が一番だと思います。
c/c++/rust/go辺りなら実用段階のコンパイラ言語なので、それなりの組み方をすればそれなりの高速化が期待できます。

[KohnoseLami]

はい、最初からわかってて質問の試したことに書いてますけどね
あなたが今このやり取りをして結論づけたことを踏まえてなにかないかを聞いていたわけなのですが、、
Processの場合は高速化には成功したが思ったより効果が出なかったと書いていますよ

私は素直にあなたの言っていることが理解出来なかったのと最初から結論は出ていてそれを踏まえてどうなのか聞いているので素直に質問をしているつもりです
結論は出ているからこそ回答者が検証する必要は無いとも言えると思っています
ダミーに関しても別に検証する必要がある質問では無い上に別にダミーくらいと仰っているように別にわざわざ結論を書いているものを検証したいならばその人が書けばと思っていますので
まずまず先に煽り口調で適当なコメント残していった方が火種だと思いますが
あれで悪意がないというのなら別ですが
少なからず私は全ての発言に悪意はありませんよ
本音を書いています
書けと仰るのならその通りにはします、ただし結論としてあなたがもうこれ以上はどうしようもないというのでしたらPythonの仕様上不可という形で自己解決します

[dameo]

結局まだ分かってないみたいですねｗ

ちゃんと書いておくと、
- そもそも質問の意味がない質問をしてる
- 事実関係の記載が皆無で他人が検証できないため、勘違いがそのままになっている
- 無責任かつワガママに他人に丸投げを意図している
- 嘘をついている(人格否定になるのでこれ以上は言わないが)
この辺が問題なのですよ。

本人的には「分かったつもりになっているが、まるで分かっていない」といったところでしょうか。
そこ私にはどうでもいいことなんですが。。。

[KohnoseLami]

私はこれ以上の詳細を求める時点で私の求めている回答を行えるレベルの人いないと思うので
この質問文で理解できない人はその程度ですよ
いちいち結論を書いているのに理解できない人がいることに理解できません
ChatGPTでも普通にこの質問が来たらほぼ1発でCythonやNumbaを進めてきてそれでもダメなら別言語っていう風に普通に誘導してくれますけどね
多分そちらこそ質問の意図がわかっていないですよ
コードの書き方の質問よりもPythonの遅さをどうにかしてくれっていう質問であることに気づくべきだと思いますが
まずその本人の結論が間違ってるんだと思われれるなら私は少なからずその人の回答は役に立たないものだと思いますね
明らかに自分を下に見てる人の回答なんて当たり前のこと言うだけなのでだいたい役に立ちませんし
初心者マークつけたらある程度マシな人が来るかと思ったんですが相変わらずこのサイトはずっとこだわりの強い人ばっかなんですね
私別に英語書けないですけど適当に翻訳かけてStackoverflowにでも投げた方が良い回答が得られそうですw

[KohnoseLami]

この質問に関しては意味が無いものと思われるのに関してはまぁ仕方ないですね
意味が無いだろうな〜、まぁなんかあったらいいな程度なので
実用段階では無いですがCodonなどの新しいものもありますしそういった埋もれているもので適したものがあればよいなと思って書いた程度なので
GPTにコードをTranslateしてもらうなんて手の回答も来るかと思ったんですけどね
質問の意図を汲み取って欲しかったです

[dameo]

やはり伝わってませんね。あなたは性格的に自分本位すぎるため、自分の間違いに自分で気付くことが出来ず、かつ他人に指摘されるのも嫌っており、事実ベースで話を進める事ができません。そのため、一生間違ったままになってしまいがちです。他人本位になれとまでは言いませんが、事実ベースで話を進めることが出来るようになれば、初心者に近づくことはできると思いますよ。技術以前にChatGPTのように全てを肯定的に受け入れ、どんなあやふやな質問でも自信を持って解をくれ、決して間違いを指摘しない相手でないと難しいとなると厳しいかもしれませんね。

質問の意図を推測させてはいけません。理由は同じ質問を知識レベルのより低い人が見て、同じように理解できないといけないからです。分かりやすい必要はありませんが、他の解釈ができないように記述してください。

SO(英語)なら適切な質問にはとても適切な回答がつきますよ。ただ変な質問は無視されるか、変な回答がつくだけです。
「意味がない」というのは、あなただけにとってではありません。全員にとって意味がないという意味です。理由は事実の記載がないために誰の参考にもならないからです。
お気持ちお察しツールのChatGPT様やその身内なら、期待に応えてくれるかもしれません。好きなだけその手のツールを使用してください。このサイトを利用する頻度も減るといいですね。

[KohnoseLami]

自分本位に関しては意識してます
煽り口調でいきなりつっかかってきておせっかい焼いてきて1回限りの会話相手なので気を使う必要は無いと思いました
やっぱり第一印象って大事だと思いますよ
あなた自身も嫌な態度で突っかかってることに気づいて直したら良いと思いますけどね
回答者は偉いみたいなの思ってます？w

他の人が同じ疑問を持って見た時にためになるようにに関してはたしかに頭から抜けてましたね〜
そこに関しては間違いなく忘れてました
回答者はレベル高い人に対等の立場で回答して欲しいので

いえ、事実Stackoverflowなどで質問したことあるので経験上のお話です
ChatGPT相手に対抗心でも燃やしてるんですか？
あれは深入りした質問にはまともな回答でも無いのに自信満々に回答するので全部信頼すべきでは無いなど使い所はわかっているつもりですので別に遠回しに言われなくても大丈夫ですw
私はそもそもこのサイト全然使っておらずたまに暇が出来たら神頼み程度で質問立てて遊んでるだけなので🥲
私はもうベストアンサーも渡してお互い好きに言い合って楽しめたのでそろそろ引かせて頂きます
おつかれでした、ありがとうございました

[dameo]

複垢さんの匂いがぷんぷんするときだけ煽り口調で様子を見ますｗ
気遣いをする人はジョーク以外事実ベースでしか話さないので、気を使う必要がありません。
相手や自分が偉いかどうかは気にしたことがないですね。
後で何か追記するかもしれませんが、ココはお開きですね。

[TakaiY]

コメントやとりされてますが長すぎて見ていないし、BA出ているのも知ってますが、コメントします。

質問者さんが解決したいのは以下のいずれですか？
- 暗号化処理そのもに時間がかかること
- 並列処理にして効率を上げる方法

[KohnoseLami]

返信遅れてしまい申し訳ございまでん
いいえ、両方とも異なります。

どちらかと言えば意見交換のカテゴリに近かったかもしれないです。
解決したいと言うよりかは暗号化処理を書き換えることが困難であり並列処理でも期待した結果が得られなかったためなにか魔法のようなものはないのか？と言った感じです

[TakaiY]

> 暗号化処理を書き換えることが困難
であれば、 暗号化処理そのもに時間がかかることは解決できないでしょう。

> 並列処理でも期待した結果が得られなかった
CPU boundの処理を、pythonのmulti threadで並列化しても効果が出ないことは回答についているとおりですが、multi processであれば短縮できる可能性はあると思います。
もし、暗号化処理そのものに時間がかかるのが問題であるなら並列化しても無意味ですが。

Answer 1

回答ではありません

質問が全然良くならないので内容をサマリして、回答を加えておきます。

質問サマリ

前提

• FastAPIを使用したREST APIを作成
• CPUがボトルネックとなる処理で、1リクエスト0.5秒程度(嘘っぽい)
• asyncを使っているがI/O waitがないので単一スレッドでは限界

試したこと

• いくつか試した(ことになってる)が、(多分やってないので)コードが出せない

質問

同時リクエストに対する処理を速くする方法はないか？

仮定

• 0.5秒は暗号化とのことでhoge.pyとして暗号化サンプルを用意したが、数MB程度では0.5秒もかからない

→ 0.5秒程度かかるループ処理を作成し、プログラム開始時に回数を測定、計測時にその回数で同様な処理を流すことで同等の負荷をCPUにかけることにした(example_async.py)

コード

venv環境に入りpython example_async.pyでwebサーバが起動すれば、http://localhost:3000/encrypt/encryptでREST APIにアクセス出来る。
別端末からsh wget.shをすれば(サーバーをバックグラウンド起動でも可)、上記URLに複数リクエストを同時に投げる(50リクエスト)ことで、捌く時間を計測できる。手元の環境だと27.5秒程度かかる。
example_async.py(環境構築スクリプトで出力されます)

from fastapi import FastAPI, Request
from pydantic import BaseModel
import time
from threading import current_thread

app = FastAPI()

def measure(func, args):
    s = time.perf_counter()
    ret = func(*args)
    e = time.perf_counter()
    return (ret, e - s)

class EncryptRequest(BaseModel):
    body: str

def f(b, count = 0, t=0.5):
    number = 0
    s = time.perf_counter()
    while True:
        is_timeup = (time.perf_counter() - s >= t)
        number += 1
        if count > 0:
            if count == number:
                return b.decode('utf8')
        elif is_timeup:
            return number

count_500ms = 0
@app.post("/encrypt/encrypt")
async def encrypt_encrypt(request: EncryptRequest):
    ret, t = measure(f, [request.body.encode('utf8'), count_500ms])
    return {"encrypted": ret, "time": t, "thread": current_thread().name, "ident": current_thread().ident}

if __name__ == "__main__":
    count_500ms = f(b'', 0, 0.5)
    print(count_500ms)
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

wget.sh(環境構築スクリプトで出力されます)

for _ in `seq 50`; do
    wget -q -O - --method=post --header 'Content-Type: application/json; charset=utf-8' --body-data='{"body":"こんにちは、世界！"}' 'http://localhost:8000/encrypt/encrypt'&
done
wait

調査

スレッドの使用

FastAPIは同期関数にすると、スレッドプールを使用するので、example_async.pyからasyncを除いただけのものです。通常の言語ではスレッドにするとCPUをコア数まで使用できるので、他に激しく動いているものがなければ、CPUxコア数近くの速度上昇が望めます。しかし、2023年4月18日現在のpythonは仕様としてnativeコードを使わない限り、スレッドによるCPU処理の並列化による高速化は期待できません。

https://docs.python.org/ja/3/glossary.html#term-global-interpreter-lock

コード

環境構築で出力されますが、僅かな変更なのでここに掲載はしません。
手元の環境では27.1秒程度かかり、asyncと変わりません(誤差程度)。
スレッドのプール数は30個。

プロセスプール(ProcessPoolExecutor)の使用

実際のCPUを使う処理を別プロセスのpythonで処理する方法です。pickleによりシリアライズしたデータ授受をQueue経由でプロセス間通信します。プロセスで並列化すればGILで頭打ちにならないため、CPUxコア数分に近い速度向上が望めます。

コード

プロセスのプール数はCPUxコア数です。
手元の環境ではCPU1個でコア数は2個(VirtualBoxで割り当てた数)で17.6秒程度。
example_async_process.py

from fastapi import FastAPI, Request
from pydantic import BaseModel
import time
from threading import current_thread
from concurrent.futures import ProcessPoolExecutor
import asyncio

app = FastAPI()

def measure(func, args):
    s = time.perf_counter()
    ret = func(*args)
    e = time.perf_counter()
    return (ret, e - s)

class EncryptRequest(BaseModel):
    body: str

def f(b, count = 0, t=0.5):
    number = 0
    s = time.perf_counter()
    while True:
        is_timeup = (time.perf_counter() - s >= t)
        number += 1
        if count > 0:
            if count == number:
                return b.decode('utf8')
        elif is_timeup:
            return number

count_500ms = 0
@app.post("/encrypt/encrypt")
async def encrypt_encrypt(request: EncryptRequest):
    loop = asyncio.get_event_loop()
    ret, t = await loop.run_in_executor(app.state.executor, measure, f, [request.body.encode('utf8'), count_500ms])
    return {"encrypted": ret, "time": t, "thread": current_thread().name, "ident": current_thread().ident}

if __name__ == "__main__":
    app.state.executor = ProcessPoolExecutor()
    count_500ms = f(b'', 0, 0.5)
    print(count_500ms)
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

回答

タイプ	ファイル名	リクエスト数	時間[s]
非同期	example_async.py	50	27.5
スレッドプール	example_thread.py	50	27.1
プロセスプール	example_async_process.py	50	17.6

CPUxコア数=2

→プロセスプールを使用すればCPUバウンドな処理でもCPUxコア数に近い速度向上が望める

環境構築スクリプト

今回は面倒なのでUnixライクなOS用しか用意していません。
空のディレクトリを作ってbashで以下を実行するだけです。venvを使っているので、中のpythonスクリプトを使う場合はenv環境に入ってから使ってください。

create_env.sh

python3 -m venv env
. env/bin/activate
pip install -U pip
pip install -U setuptools
pip install cryptography==40.0.2 fastapi==0.95.1 uvicorn==0.21.1
cat >hoge.py <<EOF
from fastapi import FastAPI, Request
from pydantic import BaseModel
from cryptography.fernet import Fernet
import time

key = Fernet.generate_key()
f = Fernet(key)
app = FastAPI()

def measure(func, args):
    s = time.perf_counter()
    ret = func(*args)
    e = time.perf_counter()
    return (ret, e - s)

@app.get("/")
async def root():
    return {"message": "Hello World"}

class EncryptRequest(BaseModel):
    body: str

@app.post("/encrypt/encrypt")
async def encrypt_encrypt(request: EncryptRequest):
    ret, t = measure(f.encrypt, [request.body.encode('utf8')])
    return {"encrypted": ret, "time": t}

@app.post("/encrypt/decrypt")
async def encrypt_decrypt(request: EncryptRequest):
    ret, t = measure(f.decrypt, [request.body.encode('utf8')])
    return {"decrypted": ret, "time": t}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)
EOF
python hoge.py&
sleep 5
for _ in `seq 10`; do
    wget -q -O - --method=post --header 'Content-Type: application/json; charset=utf-8' --body-data='{"body":"こんにちは、世界！"}' 'http://localhost:8000/encrypt/encrypt'
done
kill $(jobs -p)
wait
cat >wget.sh <<EOF
for _ in \`seq 50\`; do
    wget -q -O - --method=post --header 'Content-Type: application/json; charset=utf-8' --body-data='{"body":"こんにちは、世界！"}' 'http://localhost:8000/encrypt/encrypt'&
done
wait
EOF
cat >example_async.py <<EOF
from fastapi import FastAPI, Request
from pydantic import BaseModel
import time
from threading import current_thread

app = FastAPI()

def measure(func, args):
    s = time.perf_counter()
    ret = func(*args)
    e = time.perf_counter()
    return (ret, e - s)

class EncryptRequest(BaseModel):
    body: str

def f(b, count = 0, t=0.5):
    number = 0
    s = time.perf_counter()
    while True:
        is_timeup = (time.perf_counter() - s >= t)
        number += 1
        if count > 0:
            if count == number:
                return b.decode('utf8')
        elif is_timeup:
            return number

count_500ms = 0
@app.post("/encrypt/encrypt")
async def encrypt_encrypt(request: EncryptRequest):
    ret, t = measure(f, [request.body.encode('utf8'), count_500ms])
    return {"encrypted": ret, "time": t, "thread": current_thread().name, "ident": current_thread().ident}

if __name__ == "__main__":
    count_500ms = f(b'', 0, 0.5)
    print(count_500ms)
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)
EOF
cp -p example_async.py example_thread.py
patch -p1 <<EOF
diff -up a/example_thread.py b/example_thread.py
--- a/example_thread.py 2023-04-18 12:00:00.000000000 +0900
+++ b/example_thread.py 2023-04-18 12:00:00.000000000 +0900
@@ -28,7 +28,7 @@ def f(b, count = 0, t=0.5):
 
 count_500ms = 0
 @app.post("/encrypt/encrypt")
-async def encrypt_encrypt(request: EncryptRequest):
+def encrypt_encrypt(request: EncryptRequest):
     ret, t = measure(f, [request.body.encode('utf8'), count_500ms])
     return {"encrypted": ret, "time": t, "thread": current_thread().name, "ident": current_thread().ident}

EOF
cp -p example_async.py example_async_process.py
patch -p1 <<EOF
diff -up a/example_async_process.py b/example_async_process.py
--- a/example_async_process.py  2023-04-18 12:00:00.000000000 +0900
+++ b/example_async_process.py  2023-04-18 12:00:00.000000000 +0900
@@ -2,6 +2,8 @@ from fastapi import FastAPI, Request
 from pydantic import BaseModel
 import time
 from threading import current_thread
+from concurrent.futures import ProcessPoolExecutor
+import asyncio
 
 app = FastAPI()
 
@@ -29,10 +31,12 @@ def f(b, count = 0, t=0.5):
 count_500ms = 0
 @app.post("/encrypt/encrypt")
 async def encrypt_encrypt(request: EncryptRequest):
-    ret, t = measure(f, [request.body.encode('utf8'), count_500ms])
+    loop = asyncio.get_event_loop()
+    ret, t = await loop.run_in_executor(app.state.executor, measure, f, [request.body.encode('utf8'), count_500ms])
     return {"encrypted": ret, "time": t, "thread": current_thread().name, "ident": current_thread().ident}
 
 if __name__ == "__main__":
+    app.state.executor = ProcessPoolExecutor()
     count_500ms = f(b'', 0, 0.5)
     print(count_500ms)
     import uvicorn
EOF
for f in example_*.py; do
    echo "[$f]"
    python "$f"&
    sleep 5
    time sh wget.sh
    kill $(jobs -p)
    wait
done
deactivate

Answer 2

複数CPUを同時に使って処理したいと言うことなら、Pythonのマルチスレッドでは不可で、マルチプロセスにする必要があります。
CPUバウンドの処理であればスレッドを分ける意味は無く、分けるオーバーヘッド分だけ遅くなります。
I./Oバウンドの処理であればI/O待ちの間に別の処理が出来るので、速くなる可能性がありますが。

Comments

[KohnoseLami]

回答ありがとうございます、やはりそうなりますよね
事実Threadにすると逆効果でしたのでProcess分けするしか無いと思うのですがProcessだとProcessを作る時点で余分に使うCPUパワーが多すぎるため行う処理に対してかかるコストがあり得ないくらい高くなってしまうため難しいなと感じております。
こういった場合はやはりもう書き換えるしか無いのでしょうか？

単純に比較が出来るものではないと思いますが文字列を暗号化するだけのプログラムですが1000行近くアルゴリズムにおいてはTikTokなどで使用されているものと同等でありあまりにも書き換えるコストが高すぎるため、というよりPythonなどよりも難しい言語に書き換えるのは不可能に近くstringもしくはintegerのどちらかが引数に来る関数など動的型ならではの処理などもあるためとても難易度を高く感じます。
こういった場合Denoなどに書き換えるのが一番コスパは良いのでしょうか？

[otn]

マルチCPUで並列処理したくなるくらいの時間が掛かっているのなら、最初に考えるべきは「何処で時間が掛かっているのか」です。もしかすると実行コストの高い関数やメソッドを多用しているのかも知れません。
Pythonでやったことは無いのですが、一般的には「プロファイラー」という機能でどの行のどの関数でどの程度の時間が掛かっていたのかわかりますので、時間の掛かっている箇所を重点的に効率化検討するとかですね。
知らなかったのであれば「Python プロファイラー」で検索してみましょう。