[17.0][FIX] fastapi: Forwardport 16.0 pullrequest 486 - Avoid zombie threads

Author: siemenv

fastapi/fastapi_dispatcher.py

@@ -8,6 +8,7 @@
 
 from .context import odoo_env_ctx
 from .error_handlers import convert_exception_to_status_body
+from .pools import fastapi_app_pool
 
 
 class FastApiDispatcher(Dispatcher):
@@ -29,18 +30,17 @@ def dispatch(self, endpoint, args):
         root_path = "/" + environ["PATH_INFO"].split("/")[1]
         # TODO store the env into contextvar to be used by the odoo_env
         # depends method
-        fastapi_endpoint = self.request.env["fastapi.endpoint"].sudo()
-        app = fastapi_endpoint.get_app(root_path)
-        uid = fastapi_endpoint.get_uid(root_path)
-        data = BytesIO()
-        with self._manage_odoo_env(uid):
-            for r in app(environ, self._make_response):
-                data.write(r)
-            if self.inner_exception:
-                raise self.inner_exception
-            return self.request.make_response(
-                data.getvalue(), headers=self.headers, status=self.status
-            )
+        with fastapi_app_pool.get_app(env=request.env, root_path=root_path) as app:
+            uid = request.env["fastapi.endpoint"].sudo().get_uid(root_path)
+            data = BytesIO()
+            with self._manage_odoo_env(uid):
+                for r in app(environ, self._make_response):
+                    data.write(r)
+                if self.inner_exception:
+                    raise self.inner_exception
+                return self.request.make_response(
+                    data.getvalue(), headers=self.headers, status=self.status
+                )
 
     def handle_error(self, exc):
         headers = getattr(exc, "headers", None)

fastapi/middleware.py

@@ -0,0 +1,24 @@
+# Copyright 2025 ACSONE SA/NV
+# License LGPL-3.0 or later (http://www.gnu.org/licenses/LGPL).
+"""
+ASGI middleware for FastAPI.
+This module provides an ASGI middleware for FastAPI applications. The middleware
+is designed to ensure managed the lifecycle of the threads used to as event loop
+for the ASGI application.
+"""
+
+from collections.abc import Iterable
+
+import a2wsgi
+from a2wsgi.asgi import ASGIResponder
+from a2wsgi.wsgi_typing import Environ, StartResponse
+
+from .pools import event_loop_pool
+
+
+class ASGIMiddleware(a2wsgi.ASGIMiddleware):
+    def __call__(
+        self, environ: Environ, start_response: StartResponse
+    ) -> Iterable[bytes]:
+        with event_loop_pool.get_event_loop() as loop:
+            return ASGIResponder(self.app, loop)(environ, start_response)

fastapi/models/fastapi_endpoint.py

@@ -7,7 +7,6 @@
 from itertools import chain
 from typing import Any
 
-from a2wsgi import ASGIMiddleware
 from starlette.middleware import Middleware
 from starlette.routing import Mount
 
@@ -16,6 +15,7 @@
 from fastapi import APIRouter, Depends, FastAPI
 
 from .. import dependencies
+from ..middleware import ASGIMiddleware
 
 _logger = logging.getLogger(__name__)
 
@@ -121,10 +121,10 @@ def _registered_endpoint_rule_keys(self):
         return tuple(res)
 
     @api.model
-    def _routing_impacting_fields(self) -> tuple[str]:
+    def _routing_impacting_fields(self) -> tuple[str, ...]:
         """The list of fields requiring to refresh the mount point of the pp
         into odoo if modified"""
-        return ("root_path",)
+        return ("root_path", "save_http_session")
 
     #
     # end of endpoint.route.sync.mixin methods implementation
@@ -198,14 +198,14 @@ def _endpoint_registry_route_unique_key(self, routing: dict[str, Any]):
         return f"{self._name}:{self.id}:{path}"
 
     def _reset_app(self):
+        """When the app is reset we clear the cache, the system will signal to
+        others instances that the cache is not up to date and that they should
+        invalidate their cache as well. This is required to ensure that any change
+        requiring a reset of the app is propagated to all the running instances.
+        """
         self.env.registry.clear_cache()
 
     @api.model
-    @tools.ormcache("root_path")
-    # TODO cache on thread local by db to enable to get 1 middelware by
-    # thread when odoo runs in multi threads mode and to allows invalidate
-    # specific entries in place og the overall cache as we have to do into
-    # the _rest_app method
     def get_app(self, root_path):
         record = self.search([("root_path", "=", root_path)])
         if not record:

fastapi/pools/__init__.py

@@ -0,0 +1,11 @@
+from .event_loop import EventLoopPool
+from .fastapi_app import FastApiAppPool
+from odoo.service.server import CommonServer
+
+event_loop_pool = EventLoopPool()
+fastapi_app_pool = FastApiAppPool()
+
+
+CommonServer.on_stop(event_loop_pool.shutdown)
+
+__all__ = ["event_loop_pool", "fastapi_app_pool"]

fastapi/pools/event_loop.py

@@ -0,0 +1,61 @@
+# Copyright 2025 ACSONE SA/NV
+# License LGPL-3.0 or later (http://www.gnu.org/licenses/LGPL).
+
+import asyncio
+import queue
+import threading
+from collections.abc import Generator
+from contextlib import contextmanager
+
+
+class EventLoopPool:
+    def __init__(self):
+        self.pool = queue.Queue[tuple[asyncio.AbstractEventLoop, threading.Thread]]()
+
+    def __get_event_loop_and_thread(
+        self,
+    ) -> tuple[asyncio.AbstractEventLoop, threading.Thread]:
+        """
+        Get an event loop from the pool. If no event loop is available,
+        create a new one.
+        """
+        try:
+            return self.pool.get_nowait()
+        except queue.Empty:
+            loop = asyncio.new_event_loop()
+            thread = threading.Thread(target=loop.run_forever, daemon=True)
+            thread.start()
+            return loop, thread
+
+    def __return_event_loop(
+        self, loop: asyncio.AbstractEventLoop, thread: threading.Thread
+    ) -> None:
+        """
+        Return an event loop to the pool for reuse.
+        """
+        self.pool.put((loop, thread))
+
+    def shutdown(self):
+        """
+        Shutdown all event loop threads in the pool.
+        """
+        while not self.pool.empty():
+            loop, thread = self.pool.get_nowait()
+            loop.call_soon_threadsafe(loop.stop)
+            thread.join()
+            loop.close()
+
+    @contextmanager
+    def get_event_loop(self) -> Generator[asyncio.AbstractEventLoop, None, None]:
+        """
+        Get an event loop from the pool. If no event loop is available,
+        create a new one.
+
+        After the context manager exits, the event loop is returned to
+        the pool for reuse.
+        """
+        loop, thread = self.__get_event_loop_and_thread()
+        try:
+            yield loop
+        finally:
+            self.__return_event_loop(loop, thread)

fastapi/pools/fastapi_app.py

@@ -0,0 +1,122 @@
+# Copyright 2025 ACSONE SA/NV
+# License LGPL-3.0 or later (http://www.gnu.org/licenses/LGPL).
+import logging
+import queue
+import threading
+from collections import defaultdict
+from collections.abc import Generator
+from contextlib import contextmanager
+
+from odoo.api import Environment
+
+from fastapi import FastAPI
+
+_logger = logging.getLogger(__name__)
+
+
+class FastApiAppPool:
+    """Pool of FastAPI apps.
+    This class manages a pool of FastAPI apps. The pool is organized by database name
+    and root path. Each pool is a queue of FastAPI apps.
+    The pool is used to reuse FastAPI apps across multiple requests. This is useful
+    to avoid the overhead of creating a new FastAPI app for each request. The pool
+    ensures that only one request at a time uses an app.
+    The proper way to use the pool is to use the get_app method as a context manager.
+    This ensures that the app is returned to the pool after the context manager exits.
+    The get_app method is designed to ensure that the app made available to the
+    caller is unique and not used by another caller at the same time.
+    .. code-block:: python
+        with fastapi_app_pool.get_app(env=request.env, root_path=root_path) as app:
+            # use the app
+    The pool is invalidated when the cache registry is updated. This ensures that
+    the pool is always up-to-date with the latest app configuration. It also
+    ensures that the invalidation is done even in the case of a modification occurring
+    in a different worker process or thread or server instance. This mechanism
+    works because every time an attribute of the fastapi.endpoint model is modified
+    and this attribute is part of the list returned by the `_fastapi_app_fields`,
+    or `_routing_impacting_fields` methods, we reset the cache of a marker method
+    `_reset_app_cache_marker`. As side effect, the cache registry is marked to be
+    updated by the increment of the `cache_sequence` SQL sequence.  This cache sequence
+    on the registry is reloaded from the DB on each request made to a specific database.
+    When an app is retrieved from the pool, we always compare the cache sequence of
+    the pool with the cache sequence of the registry. If the two sequences are
+    different, we invalidate the pool and save the new cache sequence on the pool.
+    The cache is based on a defaultdict of defaultdict of queue.Queue. We are cautious
+    that the use of defaultdict is not thread-safe for operations that modify the
+    dictionary. However the only operation that modifies the dictionary is the
+    first access to a new key. If two threads access the same key at the same time,
+    the two threads will create two different queues. This is not a problem since
+    at the time of returning an app to the pool, we are sure that a queue exists
+    for the key into the cache and all the created apps are returned to the same
+    valid queue. And the end, the lack of thread-safety for the defaultdict could
+    only lead to a negligible overhead of creating a new queue that will never be
+    used. This is why we consider that the use of defaultdict is safe in this context.
+    """
+
+    def __init__(self):
+        self._queue_by_db_by_root_path: dict[
+            str, dict[str, queue.Queue[FastAPI]]
+        ] = defaultdict(lambda: defaultdict(queue.Queue))
+        self.__cache_sequence = 0
+        self._lock = threading.Lock()
+
+    def __get_pool(self, env: Environment, root_path: str) -> queue.Queue[FastAPI]:
+        db_name = env.cr.dbname
+        return self._queue_by_db_by_root_path[db_name][root_path]
+
+    def __get_app(self, env: Environment, root_path: str) -> FastAPI:
+        pool = self.__get_pool(env, root_path)
+        try:
+            return pool.get_nowait()
+        except queue.Empty:
+            return env["fastapi.endpoint"].sudo().get_app(root_path)
+
+    def __return_app(self, env: Environment, app: FastAPI, root_path: str) -> None:
+        pool = self.__get_pool(env, root_path)
+        pool.put(app)
+
+    @contextmanager
+    def get_app(
+        self, env: Environment, root_path: str
+    ) -> Generator[FastAPI, None, None]:
+        """Return a  FastAPI app to be used in a context manager.
+        The app is retrieved from the pool if available, otherwise a new one is created.
+        The app is returned to the pool after the context manager exits.
+        When used into the FastApiDispatcher class this ensures that the app is reused
+        across multiple requests but only one request at a time uses an app.
+        """
+        self._check_cache(env)
+        app = self.__get_app(env, root_path)
+        try:
+            yield app
+        finally:
+            self.__return_app(env, app, root_path)
+
+    @property
+    def cache_sequence(self) -> int:
+        return self.__cache_sequence
+
+    @cache_sequence.setter
+    def cache_sequence(self, value: int) -> None:
+        if value != self.__cache_sequence:
+            with self._lock:
+                self.__cache_sequence = value
+
+    def _check_cache(self, env: Environment) -> None:
+        cache_sequence = env.registry.cache_sequence
+        if cache_sequence != self.cache_sequence and self.cache_sequence != 0:
+            _logger.info(
+                "Cache registry updated, reset fastapi_app pool for the current "
+                "database"
+            )
+            self.invalidate(env)
+
+        self.cache_sequence = cache_sequence
+
+    def invalidate(self, env: Environment, root_path: str | None = None) -> None:
+        db_name = env.cr.dbname
+
+        if root_path:
+            self._queue_by_db_by_root_path[db_name][root_path] = queue.Queue()
+        elif db_name in self._queue_by_db_by_root_path:
+            del self._queue_by_db_by_root_path[db_name]