diff --git a/.gitignore b/.gitignore index 2c0d859ad..f405ed256 100644 --- a/.gitignore +++ b/.gitignore @@ -31,3 +31,4 @@ archive.zip .DS_Store .codspeed +.venv/ diff --git a/fastapi/routing.py b/fastapi/routing.py index c442b122b..5228c5e20 100644 --- a/fastapi/routing.py +++ b/fastapi/routing.py @@ -7,6 +7,7 @@ import inspect import json import os import stat +import threading import types from collections.abc import ( AsyncIterator, @@ -1579,49 +1580,58 @@ class _IncludedRouter(BaseRoute): default_factory=list ) _effective_low_priority_routes_version: int | None = None + _rebuild_lock: threading.Lock = field(default_factory=threading.Lock) def effective_candidates(self) -> list["_EffectiveRouteContext | _IncludedRouter"]: routes_version = self.original_router._get_routes_version() if routes_version == self._effective_candidates_version: return self._effective_candidates - self._effective_candidates = [] - candidates = self.original_router.routes - for route in candidates: - if isinstance(route, _IncludedRouter): - child_context = self.include_context.combine(route.include_context) - child_branch = _IncludedRouter( - original_router=route.original_router, - include_context=child_context, - ) - self._effective_candidates.append(child_branch) - continue - route_context = self._build_effective_context(route) - if route_context is not None: - self._effective_candidates.append(route_context) - self._effective_candidates_version = routes_version - return self._effective_candidates + with self._rebuild_lock: + if routes_version == self._effective_candidates_version: + return self._effective_candidates + new_candidates: list["_EffectiveRouteContext | _IncludedRouter"] = [] + candidates = self.original_router.routes + for route in candidates: + if isinstance(route, _IncludedRouter): + child_context = self.include_context.combine(route.include_context) + child_branch = _IncludedRouter( + original_router=route.original_router, + include_context=child_context, + ) + new_candidates.append(child_branch) + continue + route_context = self._build_effective_context(route) + if route_context is not None: + new_candidates.append(route_context) + self._effective_candidates = new_candidates + self._effective_candidates_version = routes_version + return self._effective_candidates def effective_low_priority_routes(self) -> list["_EffectiveRouteContext"]: routes_version = self.original_router._get_routes_version() if routes_version == self._effective_low_priority_routes_version: return self._effective_low_priority_routes - self._effective_low_priority_routes = [] - for route in self.original_router._low_priority_routes: - route_context = self._build_effective_context(route) - if route_context is not None: - self._effective_low_priority_routes.append(route_context) - for route in self.original_router.routes: - if isinstance(route, _IncludedRouter): - child_context = self.include_context.combine(route.include_context) - child_branch = _IncludedRouter( - original_router=route.original_router, - include_context=child_context, - ) - self._effective_low_priority_routes.extend( - child_branch.effective_low_priority_routes() - ) - self._effective_low_priority_routes_version = routes_version - return self._effective_low_priority_routes + with self._rebuild_lock: + if routes_version == self._effective_low_priority_routes_version: + return self._effective_low_priority_routes + new_low_priority: list["_EffectiveRouteContext"] = [] + for route in self.original_router._low_priority_routes: + route_context = self._build_effective_context(route) + if route_context is not None: + new_low_priority.append(route_context) + for route in self.original_router.routes: + if isinstance(route, _IncludedRouter): + child_context = self.include_context.combine(route.include_context) + child_branch = _IncludedRouter( + original_router=route.original_router, + include_context=child_context, + ) + new_low_priority.extend( + child_branch.effective_low_priority_routes() + ) + self._effective_low_priority_routes = new_low_priority + self._effective_low_priority_routes_version = routes_version + return self._effective_low_priority_routes def _build_effective_context( self, route: BaseRoute diff --git a/tests/test_thread_safety.py b/tests/test_thread_safety.py new file mode 100644 index 000000000..aef704bcf --- /dev/null +++ b/tests/test_thread_safety.py @@ -0,0 +1,95 @@ +"""Test thread safety of _IncludedRouter cache rebuild.""" +import sys +import threading + +from fastapi import APIRouter, FastAPI +from fastapi.routing import _IncludedRouter +from fastapi.testclient import TestClient + + +def test_effective_candidates_thread_safety(): + """Verify concurrent cache rebuilds don't produce duplicate candidates.""" + N_ROUTES = 120 + N_THREADS = 6 + + app = FastAPI() + router = APIRouter() + for i in range(N_ROUTES): + def endpoint(i: int = i): + return {"i": i} + router.add_api_route(f"/r{i}", endpoint, methods=["GET"]) + app.include_router(router) + + sys.setswitchinterval(1e-5) # widen the rebuild window + barrier = threading.Barrier(N_THREADS) + errors = [] + + def worker(idx: int): + try: + client = TestClient(app) + barrier.wait() + resp = client.get(f"/r{N_ROUTES - 1}") + if resp.status_code != 200: + errors.append(f"Thread {idx}: got {resp.status_code}") + except Exception as e: + errors.append(f"Thread {idx}: {e}") + + threads = [threading.Thread(target=worker, args=(i,)) for i in range(N_THREADS)] + for t in threads: + t.start() + for t in threads: + t.join() + + assert not errors, f"Errors during concurrent requests: {errors}" + + included = next( + r for r in app.router.routes if isinstance(r, _IncludedRouter) + ) + n_candidates = len(included._effective_candidates) + assert n_candidates == N_ROUTES, ( + f"Expected {N_ROUTES} cached candidates, got {n_candidates}" + ) + + +def test_effective_low_priority_thread_safety(): + """Verify concurrent low-priority route rebuilds don't duplicate.""" + N_ROUTES = 50 + N_THREADS = 4 + + app = FastAPI() + router = APIRouter() + for i in range(N_ROUTES): + def endpoint(i: int = i): + return {"i": i} + router.add_api_route(f"/r{i}", endpoint, methods=["GET"], include_in_schema=False) + app.include_router(router) + + sys.setswitchinterval(1e-5) + barrier = threading.Barrier(N_THREADS) + errors = [] + + def worker(idx: int): + try: + client = TestClient(app) + barrier.wait() + # Access low priority routes indirectly + resp = client.get("/openapi.json") + if resp.status_code != 200: + errors.append(f"Thread {idx}: got {resp.status_code}") + except Exception as e: + errors.append(f"Thread {idx}: {e}") + + threads = [threading.Thread(target=worker, args=(i,)) for i in range(N_THREADS)] + for t in threads: + t.start() + for t in threads: + t.join() + + assert not errors, f"Errors during concurrent requests: {errors}" + + +if __name__ == "__main__": + test_effective_candidates_thread_safety() + print("PASS: effective_candidates thread safety") + test_effective_low_priority_thread_safety() + print("PASS: effective_low_priority_routes thread safety")