From a4ff0719123eba34e045dd690019a4960b10c7fd Mon Sep 17 00:00:00 2001 From: oliver Date: Mon, 20 Apr 2026 17:02:23 +0100 Subject: [PATCH 1/6] =?UTF-8?q?=F0=9F=90=9B=20fix=20deadlocks=20in=20depen?= =?UTF-8?q?ds=20by=20using=20a=20separate=20CapacityLimiter=20for=20teardo?= =?UTF-8?q?wn?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit If there is a limited pool of resources, like database connections, we need separate pools of threads for acquiring and releasing those resources, otherwise there will always be deadlocks. --- fastapi/concurrency.py | 44 ++++++++++++++-------- fastapi/routing.py | 8 +++- tests/test_concurrency.py | 68 ++++++++++++++++++++++++++++++++++ tests/test_depends_deadlock.py | 56 ++++++++++++++++++++++++++++ 4 files changed, 159 insertions(+), 17 deletions(-) create mode 100644 tests/test_concurrency.py create mode 100644 tests/test_depends_deadlock.py diff --git a/fastapi/concurrency.py b/fastapi/concurrency.py index 76a5a2eb1..63a5237f4 100644 --- a/fastapi/concurrency.py +++ b/fastapi/concurrency.py @@ -1,7 +1,8 @@ -from collections.abc import AsyncGenerator +import functools +from collections.abc import AsyncGenerator, Callable from contextlib import AbstractContextManager from contextlib import asynccontextmanager as asynccontextmanager -from typing import TypeVar +from typing import ParamSpec, TypeVar import anyio.to_thread from anyio import CapacityLimiter @@ -11,31 +12,44 @@ from starlette.concurrency import ( # noqa run_until_first_complete as run_until_first_complete, ) +_P = ParamSpec("_P") _T = TypeVar("_T") +# Blocking __exit__ and other teardown operations from running can create race +# conditions/deadlocks if the context manager itself has its own internal pool +# (e.g. a database connection pool). +# To avoid this maintain a separate limiter for teardown operations, so that the +# operations acquiring resources can never block operations releasing resources. +# NOTE: 5 is arbitrary, we would like more than 1 so that teardowns are not serialised. +_teardown_limiter = CapacityLimiter(5) + @asynccontextmanager async def contextmanager_in_threadpool( cm: AbstractContextManager[_T], ) -> AsyncGenerator[_T, None]: - # blocking __exit__ from running waiting on a free thread - # can create race conditions/deadlocks if the context manager itself - # has its own internal pool (e.g. a database connection pool) - # to avoid this we let __exit__ run without a capacity limit - # since we're creating a new limiter for each call, any non-zero limit - # works (1 is arbitrary) - exit_limiter = CapacityLimiter(1) try: yield await run_in_threadpool(cm.__enter__) except Exception as e: ok = bool( - await anyio.to_thread.run_sync( - cm.__exit__, type(e), e, e.__traceback__, limiter=exit_limiter - ) + await run_in_teardown_threadpool(cm.__exit__, type(e), e, e.__traceback__) ) if not ok: raise e else: - await anyio.to_thread.run_sync( - cm.__exit__, None, None, None, limiter=exit_limiter - ) + await run_in_teardown_threadpool(cm.__exit__, None, None, None) + + +async def run_in_teardown_threadpool( + func: Callable[_P, _T], *args: _P.args, **kwargs: _P.kwargs +) -> _T: + """Run a function in the separate teardown threadpool. + + This will run the function in the teardown threadpool in order to avoid it + being blocked by other operations waiting to acquire resources. + + Unless you know what you are doing, you probably don't want this function, + use run_in_threadpool instead. + """ + func = functools.partial(func, *args, **kwargs) + return await anyio.to_thread.run_sync(func, limiter=_teardown_limiter) diff --git a/fastapi/routing.py b/fastapi/routing.py index 3a2d75422..badbe5de4 100644 --- a/fastapi/routing.py +++ b/fastapi/routing.py @@ -46,6 +46,11 @@ from fastapi._compat import ( Undefined, lenient_issubclass, ) +from fastapi.concurrency import ( + iterate_in_threadpool, + run_in_teardown_threadpool, + run_in_threadpool, +) from fastapi.datastructures import Default, DefaultPlaceholder from fastapi.dependencies.models import Dependant from fastapi.dependencies.utils import ( @@ -83,7 +88,6 @@ from fastapi.utils import ( from starlette import routing from starlette._exception_handler import wrap_app_handling_exceptions from starlette._utils import get_route_path, is_async_callable -from starlette.concurrency import iterate_in_threadpool, run_in_threadpool from starlette.datastructures import URL, FormData, URLPath from starlette.exceptions import HTTPException from starlette.requests import Request @@ -308,7 +312,7 @@ async def serialize_response( if is_coroutine: value, errors = field.validate(response_content, {}, loc=("response",)) else: - value, errors = await run_in_threadpool( + value, errors = await run_in_teardown_threadpool( field.validate, response_content, {}, loc=("response",) ) if errors: diff --git a/tests/test_concurrency.py b/tests/test_concurrency.py new file mode 100644 index 000000000..9e9fd9bff --- /dev/null +++ b/tests/test_concurrency.py @@ -0,0 +1,68 @@ +import contextlib +import time +from collections.abc import Iterator + +import anyio.to_thread +import pytest +from anyio import CapacityLimiter +from fastapi import concurrency + + +@pytest.fixture +def reset_teardown_limiter(monkeypatch: pytest.MonkeyPatch) -> None: + """Reset the teardown limiter before/after tests to avoid interference + between different anyio backends.""" + monkeypatch.setattr(concurrency, "_teardown_limiter", CapacityLimiter(5)) + + +@pytest.mark.anyio +@pytest.mark.usefixtures("reset_teardown_limiter") +async def test_run_in_teardown_threadpool() -> None: + def func(x: int, y: int) -> int: + return x + y + + result = await concurrency.run_in_teardown_threadpool(func, 1, y=2) + assert result == 3 + + +@pytest.mark.anyio +@pytest.mark.usefixtures("reset_teardown_limiter") +async def test_contextmanager_in_threadpool() -> None: + @contextlib.contextmanager + def context_manager() -> Iterator[str]: + yield "entered" + + async with concurrency.contextmanager_in_threadpool(context_manager()) as result: + assert result == "entered" + + +@pytest.mark.anyio +@pytest.mark.usefixtures("reset_teardown_limiter") +async def test_competing_acquire_release() -> None: + """Check that the main threadpool does not block the teardown threadpool.""" + pool_size = anyio.to_thread.current_default_thread_limiter().total_tokens + acquirable = False + acquired = [] + + def acquire() -> None: + while not acquirable: + time.sleep(0.001) + acquired.append(True) + + def release() -> bool: + nonlocal acquirable + time.sleep(0.001) + acquirable = True + return acquirable + + async with anyio.create_task_group() as tg: + for _ in range(pool_size): + tg.start_soon(concurrency.run_in_threadpool, acquire) + + await anyio.sleep(0.001) + + # The threadpool should now be full of threads waiting to acquire + # The release function should be able to run without being blocked by acquires + await concurrency.run_in_teardown_threadpool(release) + + assert len(acquired) == pool_size diff --git a/tests/test_depends_deadlock.py b/tests/test_depends_deadlock.py new file mode 100644 index 000000000..011cb529b --- /dev/null +++ b/tests/test_depends_deadlock.py @@ -0,0 +1,56 @@ +import asyncio +import threading +import time +from collections.abc import Iterator + +from fastapi import Depends, FastAPI +from httpx import ASGITransport, AsyncClient +from pydantic import BaseModel + +# Mutex, and dependency acting as our "connection pool" for a database for example +mutex = threading.Lock() + + +# Simulate releaasing a pooled resource in the teardown of a Depends, +# which in reality is usually a database connection or similar. +def release_resource() -> Iterator[None]: + try: + time.sleep(0.001) + yield + finally: + time.sleep(0.001) + mutex.release() + + +app = FastAPI() + + +class Item(BaseModel): + name: str + id: int + + +# An endpoint that uses Depends for resource management and also includes +# a response_model definition would previously deadlock in the validation +# of the model and the cleanup of the Depends +@app.get("/deadlock", response_model=Item) +def get_deadlock(dep: None = Depends(release_resource)) -> Item: + mutex.acquire() + return Item(name="foo", id=1) + + +# Fire off 100 requests in parallel(ish) in order to create contention +# over the shared resource (simulating a fastapi server that interacts with +# a database connection pool). +def test_depends_deadlock() -> None: + async def make_request(client: AsyncClient): + await client.get("/deadlock") + + async def run_requests() -> None: + async with AsyncClient( + transport=ASGITransport(app=app), base_url="http://testserver" + ) as aclient: + tasks = [make_request(aclient) for _ in range(100)] + await asyncio.gather(*tasks) + + asyncio.run(run_requests()) From f3a18003e66e4ae047ff1dd7bb3020931b5c78e2 Mon Sep 17 00:00:00 2001 From: oliver Date: Wed, 17 Jun 2026 21:20:22 +0100 Subject: [PATCH 2/6] =?UTF-8?q?=E2=9C=85=20Add=20timeout=20to=20anyio=20de?= =?UTF-8?q?adlock=20tests?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit This ensures that if an anyio deadlock issue is re-introduced the test suite will fail in a helpful way rather than just hanging. --- tests/test_concurrency.py | 7 +++++-- tests/test_depends_deadlock.py | 4 ++-- 2 files changed, 7 insertions(+), 4 deletions(-) diff --git a/tests/test_concurrency.py b/tests/test_concurrency.py index 9e9fd9bff..5a27e1904 100644 --- a/tests/test_concurrency.py +++ b/tests/test_concurrency.py @@ -1,3 +1,4 @@ +import asyncio import contextlib import time from collections.abc import Iterator @@ -40,7 +41,7 @@ async def test_contextmanager_in_threadpool() -> None: @pytest.mark.usefixtures("reset_teardown_limiter") async def test_competing_acquire_release() -> None: """Check that the main threadpool does not block the teardown threadpool.""" - pool_size = anyio.to_thread.current_default_thread_limiter().total_tokens + pool_size = int(anyio.to_thread.current_default_thread_limiter().total_tokens) acquirable = False acquired = [] @@ -63,6 +64,8 @@ async def test_competing_acquire_release() -> None: # The threadpool should now be full of threads waiting to acquire # The release function should be able to run without being blocked by acquires - await concurrency.run_in_teardown_threadpool(release) + await asyncio.wait_for( + concurrency.run_in_teardown_threadpool(release), timeout=10 + ) assert len(acquired) == pool_size diff --git a/tests/test_depends_deadlock.py b/tests/test_depends_deadlock.py index 011cb529b..76e44cf68 100644 --- a/tests/test_depends_deadlock.py +++ b/tests/test_depends_deadlock.py @@ -11,7 +11,7 @@ from pydantic import BaseModel mutex = threading.Lock() -# Simulate releaasing a pooled resource in the teardown of a Depends, +# Simulate releasing a pooled resource in the teardown of a Depends, # which in reality is usually a database connection or similar. def release_resource() -> Iterator[None]: try: @@ -53,4 +53,4 @@ def test_depends_deadlock() -> None: tasks = [make_request(aclient) for _ in range(100)] await asyncio.gather(*tasks) - asyncio.run(run_requests()) + asyncio.run(asyncio.wait_for(run_requests(), timeout=10)) From 79dd0ebf53dbcf3f34629382c0e367fc6afa8930 Mon Sep 17 00:00:00 2001 From: oliver Date: Sat, 20 Jun 2026 14:56:20 +0100 Subject: [PATCH 3/6] =?UTF-8?q?=E2=9C=85=20switch=20from=20asyncio.wait=5F?= =?UTF-8?q?for=20to=20pytest=20timeout=20in=20deadlock=20tests?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `wait_for` was causing the test suite to hang completely when running with `-n auto`. --- tests/test_concurrency.py | 6 ++---- tests/test_depends_deadlock.py | 4 +++- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/tests/test_concurrency.py b/tests/test_concurrency.py index 5a27e1904..9b15c8a65 100644 --- a/tests/test_concurrency.py +++ b/tests/test_concurrency.py @@ -1,4 +1,3 @@ -import asyncio import contextlib import time from collections.abc import Iterator @@ -38,6 +37,7 @@ async def test_contextmanager_in_threadpool() -> None: @pytest.mark.anyio +@pytest.mark.timeout(10, func_only=True) @pytest.mark.usefixtures("reset_teardown_limiter") async def test_competing_acquire_release() -> None: """Check that the main threadpool does not block the teardown threadpool.""" @@ -64,8 +64,6 @@ async def test_competing_acquire_release() -> None: # The threadpool should now be full of threads waiting to acquire # The release function should be able to run without being blocked by acquires - await asyncio.wait_for( - concurrency.run_in_teardown_threadpool(release), timeout=10 - ) + await concurrency.run_in_teardown_threadpool(release) assert len(acquired) == pool_size diff --git a/tests/test_depends_deadlock.py b/tests/test_depends_deadlock.py index 76e44cf68..bf9d4da9f 100644 --- a/tests/test_depends_deadlock.py +++ b/tests/test_depends_deadlock.py @@ -3,6 +3,7 @@ import threading import time from collections.abc import Iterator +import pytest from fastapi import Depends, FastAPI from httpx import ASGITransport, AsyncClient from pydantic import BaseModel @@ -42,6 +43,7 @@ def get_deadlock(dep: None = Depends(release_resource)) -> Item: # Fire off 100 requests in parallel(ish) in order to create contention # over the shared resource (simulating a fastapi server that interacts with # a database connection pool). +@pytest.mark.timeout(10, func_only=True) def test_depends_deadlock() -> None: async def make_request(client: AsyncClient): await client.get("/deadlock") @@ -53,4 +55,4 @@ def test_depends_deadlock() -> None: tasks = [make_request(aclient) for _ in range(100)] await asyncio.gather(*tasks) - asyncio.run(asyncio.wait_for(run_requests(), timeout=10)) + asyncio.run(run_requests()) From 118ffd34eca280d6623371556967ce6ea3747823 Mon Sep 17 00:00:00 2001 From: oliver Date: Sat, 20 Jun 2026 16:46:01 +0100 Subject: [PATCH 4/6] =?UTF-8?q?=E2=9C=85=20ensure=20deadlock=20tests=20pla?= =?UTF-8?q?y=20nicely=20with=20-n=20auto?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit This makes the tests more co-operative under failure, the tradeoff being that debugging may require some extra work. --- tests/test_concurrency.py | 11 +++++++---- tests/test_depends_deadlock.py | 24 ++++++++++++++++++++++-- 2 files changed, 29 insertions(+), 6 deletions(-) diff --git a/tests/test_concurrency.py b/tests/test_concurrency.py index 9b15c8a65..755bcbbf7 100644 --- a/tests/test_concurrency.py +++ b/tests/test_concurrency.py @@ -37,7 +37,6 @@ async def test_contextmanager_in_threadpool() -> None: @pytest.mark.anyio -@pytest.mark.timeout(10, func_only=True) @pytest.mark.usefixtures("reset_teardown_limiter") async def test_competing_acquire_release() -> None: """Check that the main threadpool does not block the teardown threadpool.""" @@ -46,15 +45,19 @@ async def test_competing_acquire_release() -> None: acquired = [] def acquire() -> None: - while not acquirable: + # We wait a max of 5s to acquire, which should be more than enough + for _ in range(5000): + if acquirable: + break time.sleep(0.001) + + assert acquirable, "Failed to acquire resource within timeout" acquired.append(True) - def release() -> bool: + def release() -> None: nonlocal acquirable time.sleep(0.001) acquirable = True - return acquirable async with anyio.create_task_group() as tg: for _ in range(pool_size): diff --git a/tests/test_depends_deadlock.py b/tests/test_depends_deadlock.py index bf9d4da9f..07e1a6303 100644 --- a/tests/test_depends_deadlock.py +++ b/tests/test_depends_deadlock.py @@ -1,4 +1,5 @@ import asyncio +import multiprocessing import threading import time from collections.abc import Iterator @@ -40,11 +41,30 @@ def get_deadlock(dep: None = Depends(release_resource)) -> Item: return Item(name="foo", id=1) +# NOTE: the failure mode here can cause the test suite to hang. +# Since the default threadpool is a global which is not easily patched +# (and doing so would make the test somewhat fragile/pointless) +# we need to use a subprocess to isolate the bad behaviour. +# If you are struggling to debug, comment out this function and rename +# the impl_ version to test_depends_deadlock +def test_depends_deadlock() -> None: + + proc = multiprocessing.Process(target=impl_test_depends_deadlock, daemon=True) + proc.start() + proc.join(timeout=10) + + try: + assert not proc.is_alive(), "Process is still alive after timeout, likely due to a deadlock" + finally: + proc.kill() # Ensure the process is terminated if it's still alive + + assert proc.exitcode == 0, "Process did not exit cleanly, see logs for details" + + # Fire off 100 requests in parallel(ish) in order to create contention # over the shared resource (simulating a fastapi server that interacts with # a database connection pool). -@pytest.mark.timeout(10, func_only=True) -def test_depends_deadlock() -> None: +def impl_test_depends_deadlock() -> None: async def make_request(client: AsyncClient): await client.get("/deadlock") From 51e0e5a6c9ee6f65f615bf2cb4f83beb44836f17 Mon Sep 17 00:00:00 2001 From: "pre-commit-ci-lite[bot]" <117423508+pre-commit-ci-lite[bot]@users.noreply.github.com> Date: Sat, 20 Jun 2026 15:48:04 +0000 Subject: [PATCH 5/6] =?UTF-8?q?=F0=9F=8E=A8=20Auto=20format?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tests/test_depends_deadlock.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/tests/test_depends_deadlock.py b/tests/test_depends_deadlock.py index 07e1a6303..c3d2ee075 100644 --- a/tests/test_depends_deadlock.py +++ b/tests/test_depends_deadlock.py @@ -4,7 +4,6 @@ import threading import time from collections.abc import Iterator -import pytest from fastapi import Depends, FastAPI from httpx import ASGITransport, AsyncClient from pydantic import BaseModel @@ -54,7 +53,9 @@ def test_depends_deadlock() -> None: proc.join(timeout=10) try: - assert not proc.is_alive(), "Process is still alive after timeout, likely due to a deadlock" + assert not proc.is_alive(), ( + "Process is still alive after timeout, likely due to a deadlock" + ) finally: proc.kill() # Ensure the process is terminated if it's still alive From 21eee9163f53a106734189704617542a00522f7e Mon Sep 17 00:00:00 2001 From: oliver Date: Sat, 20 Jun 2026 17:08:26 +0100 Subject: [PATCH 6/6] =?UTF-8?q?=E2=9C=85=20make=20test=5Fdepends=5Fdeadloc?= =?UTF-8?q?k=20more=20co-operative?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit This ensures that the threads which are blocking up the threadpool eventually exit when in failure mode. --- tests/test_depends_deadlock.py | 27 ++------------------------- 1 file changed, 2 insertions(+), 25 deletions(-) diff --git a/tests/test_depends_deadlock.py b/tests/test_depends_deadlock.py index c3d2ee075..2178c81f0 100644 --- a/tests/test_depends_deadlock.py +++ b/tests/test_depends_deadlock.py @@ -1,5 +1,4 @@ import asyncio -import multiprocessing import threading import time from collections.abc import Iterator @@ -36,36 +35,14 @@ class Item(BaseModel): # of the model and the cleanup of the Depends @app.get("/deadlock", response_model=Item) def get_deadlock(dep: None = Depends(release_resource)) -> Item: - mutex.acquire() + mutex.acquire(timeout=10) return Item(name="foo", id=1) -# NOTE: the failure mode here can cause the test suite to hang. -# Since the default threadpool is a global which is not easily patched -# (and doing so would make the test somewhat fragile/pointless) -# we need to use a subprocess to isolate the bad behaviour. -# If you are struggling to debug, comment out this function and rename -# the impl_ version to test_depends_deadlock -def test_depends_deadlock() -> None: - - proc = multiprocessing.Process(target=impl_test_depends_deadlock, daemon=True) - proc.start() - proc.join(timeout=10) - - try: - assert not proc.is_alive(), ( - "Process is still alive after timeout, likely due to a deadlock" - ) - finally: - proc.kill() # Ensure the process is terminated if it's still alive - - assert proc.exitcode == 0, "Process did not exit cleanly, see logs for details" - - # Fire off 100 requests in parallel(ish) in order to create contention # over the shared resource (simulating a fastapi server that interacts with # a database connection pool). -def impl_test_depends_deadlock() -> None: +def test_depends_deadlock() -> None: async def make_request(client: AsyncClient): await client.get("/deadlock")