Sync endpoints dont finish requests before accepting a next one in high load

[LukasJerabek]

First Check

• I added a very descriptive title here.
• I used the GitHub search to find a similar question and didn't find it.
• I searched the FastAPI documentation, with the integrated search.
• I already searched in Google "How to X in FastAPI" and didn't find any information.
• I already read and followed all the tutorial in the docs and didn't find an answer.
• I already checked if it is not related to FastAPI but to Pydantic.
• I already checked if it is not related to FastAPI but to Swagger UI.
• I already checked if it is not related to FastAPI but to ReDoc.

Commit to Help

• I commit to help with one of those options 👆

Example Code

from contextlib import asynccontextmanager
from typing import Dict, Optional

from fastapi import FastAPI, HTTPException, status
from sqlalchemy import MetaData, Table, create_engine, select
from sqlalchemy.engine import Connection, Engine
from sqlalchemy.pool import QueuePool

import shared_globals.config as c


class Database:
    def __init__(self, engine: Engine, metadata: MetaData):
        self.engine = engine
        self.metadata = metadata

    @classmethod
    def init(cls, conn_str: str, metadata: Optional[MetaData] = None, **kwargs):
        engine_config = Database.get_engine_config(**kwargs)
        engine = create_engine(conn_str, **engine_config)

        if metadata is None:
            metadata = MetaData()
            metadata.reflect(bind=engine, views=True)

        return cls(engine=engine, metadata=metadata)

    @staticmethod
    def get_engine_config(
        echo: bool = False,
        future: bool = False,
        poolclass=QueuePool,
        pool_pre_ping: bool = True,
        **kwargs,
    ):
        kwargs["echo"] = echo
        kwargs["future"] = future
        kwargs["poolclass"] = poolclass
        kwargs["pool_pre_ping"] = pool_pre_ping

        return kwargs

    def create_tables(self):
        self.metadata.create_all(bind=self.engine)

    def connect(self):
        connection = self.engine.connect()
        return DatabaseConnectionManager(
            engine=self.engine, metadata=self.metadata, connection=connection
        )

    @property
    def tables(self):
        return self.metadata.tables


class DatabaseConnectionManager:
    def __init__(self, engine: Engine, metadata: MetaData, connection: Connection):
        self.engine = engine
        self.metadata = metadata
        self.connection: Connection = connection

    def commit(self):
        self.connection.commit()

    def rollback(self):
        self.connection.rollback()

    def close(self):
        self.connection.close()

    def execute(self, *args, **kwargs):
        return self.connection.execute(*args, **kwargs)

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_trace):
        self.close()

    @property
    def tables(self) -> Dict[str, Table]:
        return self.metadata.tables

    def table(self, table_name: str) -> Table:
        return Table(table_name, self.metadata, autoload_with=self.engine)


class SubjectStorage:
    def __init__(self, sync_db: DatabaseConnectionManager) -> None:
        self.sync_db = sync_db

    def sync_get_subject_by_id(self, subject_id: int, request_no: int):
        print(f"{request_no} start storage")
        subject = self.sync_db.tables["subject"]
        print(f"{request_no} start storage await")
        result = self.sync_db.connection.execute(
            select(subject.c.id).where(subject.c.id == subject_id)
        )
        print(f"{request_no} end storage await")

        rows = result.fetchall()

        if not rows:
            raise LookupError(f"Subject with ID {subject_id} not found.")

        print(f"{request_no} end storage")
        return rows


sync_ia_database = Database.init(
    c.IA_SQLALCHEMY_DB_CONN_STR,
    future=True,
    pool_size=100,
    connect_args={"options": "-c statement_timeout=600000"},  # miliseconds
    pool_timeout=600,
)


def sync_get_subject_by_id(
    subject_storage: SubjectStorage,
    subject_id: int,
    request_no: int,
):
    print(f"{request_no} start usecase")
    result = subject_storage.sync_get_subject_by_id(
        subject_id=subject_id, request_no=request_no
    )
    print(f"{request_no} end usecase")
    return result


@asynccontextmanager
async def lifespan(app: FastAPI):
    yield
    sync_ia_database.engine.dispose()


app = FastAPI(title="Test", lifespan=lifespan, debug=True)


@app.get(
    "/sync/{subject_id}/{request_no}",
    status_code=status.HTTP_200_OK,
)
def sync_get_single_subject(
    subject_id: int,
    request_no: int,
):
    with sync_ia_database.connect() as db:
        print(f"{request_no} start")
        subject_storage = SubjectStorage(sync_db=db)
        try:
            return sync_get_subject_by_id(
                subject_storage=subject_storage,
                subject_id=subject_id,
                request_no=request_no,
            )
        except LookupError as err:
            print(f"{request_no} error")
            raise HTTPException(status.HTTP_404_NOT_FOUND, detail="Not Found") from err
        finally:
            print(f"{request_no} end")

Description

I have two VMs

# -*- mode: ruby -*-
# vi: set ft=ruby :

VB_MEMORY = 4096
VB_CPUS = 1
Vagrant.configure("2") do |config|
  config.vm.box = "generic/centos7"
  config.vm.box_check_update = true
  config.vm.box_version = "4.3.12"
  config.vm.hostname = "devcentostest"
  config.vm.network "private_network", ip: "192.168.56.112"
  config.vm.define "devcentostest"
  config.vm.provider "virtualbox" do |vb|
    vb.name = "devcentostest"
    vb.gui = false
    vb.memory = VB_MEMORY
    vb.cpus = VB_CPUS
    vb.customize [ "guestproperty", "set", :id, "/VirtualBox/GuestAdd/VBoxService/--timesync-set-threshold", 1000 ]
  end
  config.vm.network :forwarded_port, guest: 22, host: 2224, id: 'ssh'
  config.ssh.extra_args = ["-L", "8001:localhost:8001"]
end

and

# -*- mode: ruby -*-
# vi: set ft=ruby :

VB_MEMORY = 4096
VB_CPUS = 1
Vagrant.configure("2") do |config|
  config.vm.box = "generic/centos7"
  config.vm.box_check_update = true
  config.vm.box_version = "4.3.12"
  config.vm.hostname = "devcentos7"
  config.vm.network "private_network", ip: "192.168.56.110"
  config.vm.define "devcentos7"
  config.vm.provider "virtualbox" do |vb|
    vb.name = "devcentos7"
    vb.gui = false
    vb.memory = VB_MEMORY
    vb.cpus = VB_CPUS
    vb.customize [ "guestproperty", "set", :id, "/VirtualBox/GuestAdd/VBoxService/--timesync-set-threshold", 1000 ]
  end
  config.vm.network :forwarded_port, guest: 22, host: 2222, id: 'ssh'
  config.ssh.extra_args = ["-L", "8000:localhost:8000"]
end

first is hosting fastapi's above code run by (install necessary requirements in the venv):
PYTHONPATH=$PYTHONPATH:~/pack:~/pack/web2py ~/venv/pack/bin/gunicorn --config /home/vagrant/pack/iadmin_fastapi/modules/resources/test_sync.conf.py --timeout 120 -b 127.0.0.1:8001

sync.conf.py:

loglevel = "debug"
wsgi_app = "test_sync:app"
pidfile = "/home/vagrant/sync-asgi-test.pid"
proc_name = "sync-asgi-test"
chdir = "/home/vagrant/"
workers = 1
worker_class = "uvicorn.workers.UvicornWorker"
proc_name = "sync process"

Now the other VM is hosting a postgresql with one table subject with column id and name. It cant be quick query, so that the db was blocking a little big. In my case 1000 000 of records without index made it take 200 ms each search by id.

CREATE TABLE public.subject (
    id integer NOT NULL,
    name text NOT NULL,
);

CREATE SEQUENCE public.subject_id_seq
    START WITH 1
    INCREMENT BY 1
    NO MINVALUE
    NO MAXVALUE
    CACHE 1;

ALTER SEQUENCE public.subject_id_seq OWNED BY public.subject.id;

DO $$
BEGIN
FOR i IN 1..1000000 LOOP
	INSERT INTO subject (name)
	VALUES (
		'a'
	);
END LOOP;
END
$$

You also have to set postgresql.conf max_connections to 10 000 and restart, so that its not blocking the test.

Now in my local server thats hosting these 2 VMs, I run this script

import asyncio
import sys
import time
import traceback
from types import TracebackType
from typing import Tuple, Type, Union

import aiohttp
from typing_extensions import TypeAlias

CONNECTION_TIMEOUT_SECONDS = 600
READ_TIMEOUT_SECONDS = 600

timeout = aiohttp.ClientTimeout(
    connect=CONNECTION_TIMEOUT_SECONDS, sock_read=READ_TIMEOUT_SECONDS
)


ExcInfo: TypeAlias = Tuple[Type[BaseException], BaseException, TracebackType]
OptExcInfo: TypeAlias = Union[ExcInfo, Tuple[None, None, None]]


async def run_task_with_full_error_info(task):
    return await task


async def make_request(
    session: aiohttp.ClientSession, url: str, request_no: int
) -> Union[OptExcInfo, aiohttp.ClientResponse]:
    formatted_url = url.format(request_no=request_no)
    print(formatted_url)
    async with session.get(formatted_url, timeout=timeout) as response:
        print(f"odpověď obdržena request {request_no}")
        try:
            await response.text()
        except Exception:
            return sys.exc_info()
        return response


async def run_test(url, num_requests):
    conn = aiohttp.TCPConnector(limit=0)
    async with aiohttp.ClientSession(connector=conn) as session:
        print("tasks_not_ready")
        tasks = [
            run_task_with_full_error_info(make_request(session, url, request_no))
            for request_no in range(num_requests)
        ]
        print("tasks_ready")
        start_time = time.time()
        print("pred gather")
        responses = await asyncio.gather(*tasks, return_exceptions=True)
        print("po gather")
        end_time = time.time()

    success_count = 0
    for response in responses:
        print("response")
        if isinstance(response, aiohttp.ClientResponse):
            try:
                response.raise_for_status()
            except Exception as err:
                print(err)
            print(response.status)
            if response.status == 200:
                success_count += 1
        else:
            print("error")
            print(response)
            if not isinstance(response, BaseException):
                exc_type, exc_value, exc_traceback = response
                print(f"Task failed with exception:")
                print(
                    "".join(
                        traceback.format_exception(exc_type, exc_value, exc_traceback)
                    )
                )
            else:
                print("neni")

    total_time = end_time - start_time
    requests_per_second = num_requests / total_time

    print(f"Total requests: {num_requests}")
    print(f"Successful requests: {success_count}")
    print(f"Total time: {total_time:.2f} seconds")
    print(f"Requests per second: {requests_per_second:.2f}")


if __name__ == "__main__":

    suffix = "115111/{request_no}"
    TEST_URL = f"http://127.0.0.1:8001/sync/{suffix}"
    NUM_REQUESTS = 1000  # Adjust as needed

    asyncio.run(run_test(TEST_URL, NUM_REQUESTS))

This makes 1000 requests at a time.

The problem is my observation:

• when I do 1000 requests at first I see like 6 responses getting back to the script I do requests with. Then it holds for a while and then I get all the responses at once. This could be a problem if requests kept comming. However in logging on the server side I see that whole code of the endpoint was finished. It seems like the response would hold back even though the data is already retrieved from the database. Like the incomming requests would jam the ability to respond from already finished threads. This for me seem unexpected.

Expected behaviour:

• already finished requests would return response immediately once they are finished and wont be waiting for other incomming requests to finish as well.

Operating System

Linux

Operating System Details

Centos7 (see VMs above)

FastAPI Version

0.111.0

Pydantic Version

2.7.1

Python Version

3.8.19

Additional Context

As you can see in fastapi code I am using with statement for connections in the endpoint. At first I was using dependency:

def get_ia_db_connection():
    conn = ia_database.connect()
    try:
        yield conn
    finally:
        await conn.close()

    sync_db: Annotated[
        DatabaseConnectionManager, Depends(get_ia_db_connection)
    ],

However it wouldnt close my connections ever and the whole code would not process not a single request and the client would time out eventually.

This I feel is connected. However I then read somewhere that the sync dependency exit part is evaluated in separate thread and that would make sense then why it doesnt work. But what would I think is that once I dont use dependency, the thread would finish without a problem and wont hold.

[LukasJerabek]

Ill just mention that async endpoint of the same test works flawlessly, also sync endpoint with query that is very fast and pool_size is set to 10 000 instead of 100 also works well. The problems are only with sync endpoint and at least a little bit blocking IO operation.

Feel free to ask if any futher explanation or simplification is needed...

[LukasJerabek]

@tiangolo can we make an issue out of this one?

[YuriiMotov]

I would recommend you simplify your code example as much as possible. Ideally write code without external dependencies (database, ...).
In that case there will be more chances that somebody take a look and help you

[LukasJerabek]

I would recommend you simplify your code example as much as possible. Ideally write code without external dependencies (database, ...). In that case there will be more chances that somebody take a look and help you

@YuriiMotov So i dug into simplifying and I noticed what bypasses the problem.

Once I remove annotation of the return value of the endpoint in this sync version, suddenly I get responses continuously as expected. Out of your mind, are you aware of such problem?

When I have time, I have to check if the issue is already created somewhere here and then maybe rephrase whole problem in another discussion.