FastAPIでスレッドエラーとなってしまう

Question

前提・実現したいこと

FastAPIにてREST APIを作成し、サーバ(Conoha Wing)に乗せてlocalhostで動かし、
.htaccessにてリバースプロキシさせて稼働しています。(フロントはReact)
FastAPIの起動には下記設定でgunicornを利用しています。
■ gunicorn設定ファイル

name = 'gunicorn'

worker_class = 'uvicorn.workers.UvicornWorker'
workers = 5
threads = 5
keepalive = 2

起動後しばらくは良いのですが、
何度かアクセスを行うとスレッド作成ができないエラーが発生してしまいます。

実際にps -fLu [ユーザID]でプロセスを確認してみると、
起動直後はプロセスは6つ(親 + worker 5個)ですが、
アクセス毎にこのプロセスが増えていきどこかのタイミングで上限に達してしまっているようです。

このプロセスは本来はアクセス毎に生成され、通信が終了した時点で消えるものなのではないのでしょうか？
また、このエラーを解決する方法はありますでしょうか？

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

  File ".../api/venv/lib64/python3.6/site-packages/fastapi/dependencies/utils.py", line 550, in solve_dependencies
    solved = await run_in_threadpool(call, **sub_values)
  File ".../api/venv/lib64/python3.6/site-packages/starlette/concurrency.py", line 40, in run_in_threadpool
    return await loop.run_in_executor(None, func, *args)
  File "uvloop/loop.pyx", line 2658, in uvloop.loop.Loop.run_in_executor
  File "/opt/alt/python36/lib64/python3.6/concurrent/futures/thread.py", line 123, in submit
    self._adjust_thread_count()
  File "/opt/alt/python36/lib64/python3.6/concurrent/futures/thread.py", line 142, in _adjust_thread_count
    t.start()
  File "/opt/alt/python36/lib64/python3.6/threading.py", line 846, in start
    _start_new_thread(self._bootstrap, ())
RuntimeError: can't start new thread

試したこと

通信を受けるメソッドをasyncで定義すると解決するとあったのですがそれでもだめでした。

@router.get("/", response_model=List[schemas.User])
async def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = crud_user.get_users(db, skip=skip, limit=limit)
    return users

下記のように構成を小さくして再度試してみました。
結果的にはやはりプロセスが残っており、上限?の70プロセスで毎回エラーとなります。
ただし、DBアクセスを行わない'/'へのアクセスではプロセスの残りはなかったので、
DB周りが怪しいということが分かりました。closeはちゃんとしているんですが。。。
■ FastAPI側

# main.py
from fastapi import Depends, FastAPI, Request, Response
from fastapi.middleware.cors import CORSMiddleware
from sqlalchemy.orm import Session
from typing import List

from database import models, schemas
from database.database import SessionLocal, engine
from database.cruds import user as crud_user
from dependencies import get_db

models.Base.metadata.create_all(bind=engine)

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


@app.middleware('http')
async def db_session_middleware(request: Request, call_next):
    response = Response('Internal server error', status_code=500)
    try:
        request.state.db = SessionLocal()
        response = await call_next(request)
    finally:
        request.state.db.close()

    return response


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


@app.get("/users", response_model=List[schemas.User])
async def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = crud_user.get_users(db, skip=skip, limit=limit)
    return users

# dependencies.py
from fastapi import Request


def get_db(request: Request):
    return request.state.db

# database/database.py
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

import json

with open('config.json', mode='r') as f:
    config = json.load(f)
    database_info = config['database']
    DATABASE_URL = "mysql+pymysql://%s:%s@%s/%s?charset=utf8mb4" % (
        database_info['user'],
        database_info['password'],
        database_info['host'],
        database_info['db'],
    )

engine = create_engine(DATABASE_URL, echo=False, pool_recycle=60)

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

# database/cruds/user.py
from sqlalchemy.orm import Session

from database import models

from typing import List


def get_users(db: Session, skip: int = 0, limit: int = 100) -> List[models.User]:
    return db.query(models.User).offset(skip).limit(limit).all()

■ API実行用
テスト用に作ったもので実際はReactからaxiosで実行しています。

import subprocess


def main():
    for i in range(0, 100):
        print(f'{i+1}回目')
        subprocess.call('curl http://localhost:8001/users', shell=True)


if __name__ == '__main__':
    main()

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

・python => 3.6
・FastAPI => 0.65.1
・uvicorn => 0.13.4
・gunicorn => 20.1.0

サーバ
・Conoha Wing

Answer 1

自己解決しました。
やはりDB周りだったようです。

# dependencies.py
from fastapi import Request

# def get_db(request: Request):      <- old
async def get_db(request: Request): #<- new
    return request.state.db

FastAPIのチュートリアルにはoldで書かれているので注意が必要そうですね。

Comments

[m2l]

解決できて良かったです！

Answer 2

fastapi側の処理としてどのような形になっていますでしょうか？
レスポンスを返さず保持した状態ですとthreadがどんどん増えていきまして、スレッドエラーになるかと思います。
お手数でなければ該当のurlに対しての処理が書かれたコードをご提示頂くことは可能でしょうか？

Comments

[p19ljk]

コメントありがとうございます。

構成を小さくして再度テストしてみたものを追記いたしました。
urlに対しての処理はmain.pyのread_usersになります。