Allow caching of middleware construction (aio-libs#10890)

Author: Dreamsorcerer

CHANGES/10890.misc.rst

@@ -0,0 +1 @@
+Improved client middleware performance by allowing the chain of middlewares to be cached across requests -- by :user:`Dreamsorcerer`.

aiohttp/client.py

@@ -68,13 +68,15 @@
     WSMessageTypeError,
     WSServerHandshakeError,
 )
-from .client_middlewares import ClientMiddlewareType, build_client_middlewares
+from .client_middlewares import ClientMiddlewareType, _cached_build_client_middlewares
 from .client_reqrep import (
     SSL_ALLOWED_TYPES,
     ClientRequest,
     ClientResponse,
+    ClientTimeout,
     Fingerprint,
     RequestInfo,
+    ResponseParams,
 )
 from .client_ws import (
     DEFAULT_WS_CLIENT_TIMEOUT,
@@ -90,7 +92,6 @@
     HTTP_AND_EMPTY_SCHEMA_SET,
     TimeoutHandle,
     _auth_header_from_netrc,
-    frozen_dataclass_decorator,
     get_env_proxy_for_url,
     netrc_from_env,
     sentinel,
@@ -214,52 +215,6 @@ class _WSConnectOptions(TypedDict, total=False):
     max_msg_size: int
 
 
-@frozen_dataclass_decorator
-class ClientTimeout:
-    total: float | None = 5 * 60  # 5 minute default timeout
-    connect: float | None = None
-    sock_read: float | None = None
-    sock_connect: float | None = None
-    ceil_threshold: float = 5
-
-    # pool_queue_timeout: Optional[float] = None
-    # dns_resolution_timeout: Optional[float] = None
-    # socket_connect_timeout: Optional[float] = None
-    # connection_acquiring_timeout: Optional[float] = None
-    # new_connection_timeout: Optional[float] = None
-    # http_header_timeout: Optional[float] = None
-    # response_body_timeout: Optional[float] = None
-
-    # to create a timeout specific for a single request, either
-    # - create a completely new one to overwrite the default
-    # - or use https://docs.python.org/3/library/dataclasses.html#dataclasses.replace
-    # to overwrite the defaults
-
-    def __post_init__(self) -> None:
-        # Ensure total is never lower than a more specific timeout, otherwise
-        # the latter would be silently capped by total and rendered useless.
-        # total=None means the user explicitly disabled the total timeout.
-        if self.total is None:
-            return
-        object.__setattr__(
-            self,
-            "total",
-            max(
-                self.total,
-                self.connect or 0,
-                self.sock_read or 0,
-                self.sock_connect or 0,
-            ),
-        )
-
-        if self.total == 0:
-            raise ValueError(
-                "total timeout must be a positive number or None to disable, "
-                "got 0. Using 0 to disable timeouts is no longer supported, "
-                "use None instead."
-            )
-
-
 # https://www.rfc-editor.org/rfc/rfc9110#section-9.2.2
 IDEMPOTENT_METHODS = frozenset({"GET", "HEAD", "OPTIONS", "TRACE", "PUT", "DELETE"})
 
@@ -271,6 +226,31 @@ def __post_init__(self) -> None:
 _CharsetResolver = Callable[[ClientResponse, bytes], str]
 
 
+# Module-level (not a closure) so it has a stable identity for the
+# ``_cached_build_client_middlewares`` cache key.
+async def _connect_and_send_request(req: ClientRequest) -> ClientResponse:
+    connector = req._session._connector
+    assert connector is not None
+    try:
+        conn = await connector.connect(req, traces=req._traces, timeout=req._timeout)
+    except asyncio.TimeoutError as exc:
+        raise ConnectionTimeoutError(f"Connection timeout to host {req.url}") from exc
+
+    assert conn.protocol is not None
+    conn.protocol.set_response_params(**req._response_params)
+    try:
+        resp = await req._send(conn)
+        try:
+            await resp.start(conn)
+        except BaseException:
+            resp.close()
+            raise
+    except BaseException:
+        conn.close()
+        raise
+    return resp
+
+
 @final
 class ClientSession:
     """First-class interface for making HTTP requests."""
@@ -428,7 +408,7 @@ def __init__(
 
         self._default_proxy = proxy
         self._retry_connection: bool = True
-        self._middlewares = middlewares
+        self._middlewares = tuple(middlewares)
 
     def __init_subclass__(cls: type["ClientSession"]) -> None:
         raise TypeError(
@@ -674,6 +654,19 @@ async def _request(
                                     {hdrs.PROXY_AUTHORIZATION: env_proxy_auth}
                                 )
 
+                    response_params: ResponseParams = {
+                        "timer": timer,
+                        "skip_payload": method in EMPTY_BODY_METHODS,
+                        "read_until_eof": read_until_eof,
+                        "auto_decompress": auto_decompress,
+                        "read_timeout": real_timeout.sock_read,
+                        "read_bufsize": read_bufsize,
+                        "timeout_ceil_threshold": self._connector._timeout_ceil_threshold,
+                        "max_line_size": max_line_size,
+                        "max_field_size": max_field_size,
+                        "max_headers": max_headers,
+                    }
+
                     req = self._request_class(
                         method,
                         url,
@@ -689,7 +682,9 @@ async def _request(
                         loop=self._loop,
                         response_class=self._response_class,
                         proxy=proxy_,
+                        response_params=response_params,
                         timer=timer,
+                        timeout=real_timeout,
                         session=self,
                         ssl=ssl,
                         server_hostname=server_hostname,
@@ -698,52 +693,13 @@ async def _request(
                         trust_env=self.trust_env,
                     )
 
-                    async def _connect_and_send_request(
-                        req: ClientRequest,
-                    ) -> ClientResponse:
-                        # connection timeout
-                        assert self._connector is not None
-                        try:
-                            conn = await self._connector.connect(
-                                req, traces=traces, timeout=real_timeout
-                            )
-                        except asyncio.TimeoutError as exc:
-                            raise ConnectionTimeoutError(
-                                f"Connection timeout to host {req.url}"
-                            ) from exc
-
-                        assert conn.protocol is not None
-                        conn.protocol.set_response_params(
-                            timer=timer,
-                            skip_payload=req.method in EMPTY_BODY_METHODS,
-                            read_until_eof=read_until_eof,
-                            auto_decompress=auto_decompress,
-                            read_timeout=real_timeout.sock_read,
-                            read_bufsize=read_bufsize,
-                            timeout_ceil_threshold=self._connector._timeout_ceil_threshold,
-                            max_line_size=max_line_size,
-                            max_field_size=max_field_size,
-                            max_headers=max_headers,
-                        )
-                        try:
-                            resp = await req._send(conn)
-                            try:
-                                await resp.start(conn)
-                            except BaseException:
-                                resp.close()
-                                raise
-                        except BaseException:
-                            conn.close()
-                            raise
-                        return resp
-
                     # Apply middleware (if any) - per-request middleware overrides session middleware
                     effective_middlewares = (
-                        self._middlewares if middlewares is None else middlewares
+                        self._middlewares if middlewares is None else tuple(middlewares)
                     )
 
                     if effective_middlewares:
-                        handler = build_client_middlewares(
+                        handler = _cached_build_client_middlewares(
                             _connect_and_send_request, effective_middlewares
                         )
                     else:

aiohttp/client_middlewares.py

@@ -1,6 +1,7 @@
 """Client middleware support."""
 
-from collections.abc import Awaitable, Callable, Sequence
+from collections.abc import Awaitable, Callable
+from functools import lru_cache
 
 from .client_reqrep import ClientRequest, ClientResponse
 
@@ -17,16 +18,15 @@
 
 def build_client_middlewares(
     handler: ClientHandlerType,
-    middlewares: Sequence[ClientMiddlewareType],
+    middlewares: tuple[ClientMiddlewareType, ...],
 ) -> ClientHandlerType:
     """
     Apply middlewares to request handler.
 
     The middlewares are applied in reverse order, so the first middleware
     in the list wraps all subsequent middlewares and the handler.
 
-    This implementation avoids using partial/update_wrapper to minimize overhead
-    and doesn't cache to avoid holding references to stateful middleware.
+    This implementation avoids using partial/update_wrapper to minimize overhead.
     """
     # Optimize for single middleware case
     if len(middlewares) == 1:
@@ -53,3 +53,6 @@ async def wrapped(req: ClientRequest) -> ClientResponse:
         current_handler = make_wrapper(middleware, current_handler)
 
     return current_handler
+
+
+_cached_build_client_middlewares = lru_cache(maxsize=64)(build_client_middlewares)

aiohttp/client_reqrep.py

@@ -83,6 +83,52 @@
 _DIGITS_RE = re.compile(r"\d+", re.ASCII)
 
 
+@frozen_dataclass_decorator
+class ClientTimeout:
+    total: float | None = 5 * 60  # 5 minute default timeout
+    connect: float | None = None
+    sock_read: float | None = None
+    sock_connect: float | None = None
+    ceil_threshold: float = 5
+
+    # pool_queue_timeout: Optional[float] = None
+    # dns_resolution_timeout: Optional[float] = None
+    # socket_connect_timeout: Optional[float] = None
+    # connection_acquiring_timeout: Optional[float] = None
+    # new_connection_timeout: Optional[float] = None
+    # http_header_timeout: Optional[float] = None
+    # response_body_timeout: Optional[float] = None
+
+    # to create a timeout specific for a single request, either
+    # - create a completely new one to overwrite the default
+    # - or use https://docs.python.org/3/library/dataclasses.html#dataclasses.replace
+    # to overwrite the defaults
+
+    def __post_init__(self) -> None:
+        # Ensure total is never lower than a more specific timeout, otherwise
+        # the latter would be silently capped by total and rendered useless.
+        # total=None means the user explicitly disabled the total timeout.
+        if self.total is None:
+            return
+        object.__setattr__(
+            self,
+            "total",
+            max(
+                self.total,
+                self.connect or 0,
+                self.sock_read or 0,
+                self.sock_connect or 0,
+            ),
+        )
+
+        if self.total == 0:
+            raise ValueError(
+                "total timeout must be a positive number or None to disable, "
+                "got 0. Using 0 to disable timeouts is no longer supported, "
+                "use None instead."
+            )
+
+
 def _gen_default_accept_encoding() -> str:
     encodings = [
         "gzip",
@@ -184,6 +230,19 @@ class ConnectionKey(NamedTuple):
     server_hostname: str | None = None
 
 
+class ResponseParams(TypedDict):
+    timer: BaseTimerContext | None
+    skip_payload: bool
+    read_until_eof: bool
+    auto_decompress: bool
+    read_timeout: float | None
+    read_bufsize: int
+    timeout_ceil_threshold: float
+    max_line_size: int
+    max_field_size: int
+    max_headers: int
+
+
 class ClientResponse(HeadersMixin):
     # Some of these attributes are None when created,
     # but will be set by the start() method.
@@ -955,7 +1014,9 @@ class ClientRequestArgs(TypedDict, total=False):
     loop: asyncio.AbstractEventLoop
     response_class: type[ClientResponse]
     proxy: URL | None
+    response_params: ResponseParams
     timer: BaseTimerContext
+    timeout: ClientTimeout
     session: "ClientSession"
     ssl: SSLContext | bool | Fingerprint
     proxy_headers: CIMultiDict[str] | None
@@ -968,6 +1029,10 @@ class ClientRequest(ClientRequestBase):
     _EMPTY_BODY = payload.PAYLOAD_REGISTRY.get(b"", disposition=None)
     _body = _EMPTY_BODY
     _continue = None  # waiter future for '100 Continue' response
+    _response_params: ResponseParams = None  # type: ignore[assignment]
+    _session: "ClientSession" = None  # type: ignore[assignment]
+    _timeout = ClientTimeout()
+    _traces: list["Trace"] = ()  # type: ignore[assignment]
 
     GET_METHODS = {
         hdrs.METH_GET,
@@ -997,7 +1062,9 @@ def __init__(
         loop: asyncio.AbstractEventLoop,
         response_class: type[ClientResponse],
         proxy: URL | None,
+        response_params: ResponseParams,
         timer: BaseTimerContext,
+        timeout: ClientTimeout,
         session: "ClientSession",
         ssl: SSLContext | bool | Fingerprint,
         proxy_headers: CIMultiDict[str] | None,
@@ -1021,7 +1088,9 @@ def __init__(
         self._session = session
         self.chunked = chunked
         self.response_class = response_class
+        self._response_params = response_params
         self._timer = timer
+        self._timeout = timeout
         self.server_hostname = server_hostname
         self.version = version
 

tests/conftest.py

@@ -32,7 +32,7 @@
 except ImportError:  # For downstreams only  # pragma: no cover
     HAS_BLOCKBUSTER = False
 
-from aiohttp.client import ClientSession
+from aiohttp.client import ClientSession, ClientTimeout
 from aiohttp.client_proto import ResponseHandler
 from aiohttp.client_reqrep import ClientRequest, ClientRequestArgs, ClientResponse
 from aiohttp.compression_utils import ZLibBackend, ZLibBackendProtocol, set_zlib_backend
@@ -439,6 +439,8 @@ def maker(
     ) -> ClientRequest:
         session = ClientSession()
         sessions.append(session)
+        timer = TimerNoop()
+        timeout = ClientTimeout()
         default_args: ClientRequestArgs = {
             "loop": asyncio.get_running_loop(),
             "params": {},
@@ -452,7 +454,20 @@ def maker(
             "expect100": False,
             "response_class": ClientResponse,
             "proxy": None,
-            "timer": TimerNoop(),
+            "response_params": {
+                "timer": timer,
+                "skip_payload": True,
+                "read_until_eof": True,
+                "auto_decompress": True,
+                "read_timeout": timeout.sock_read,
+                "read_bufsize": 2**16,
+                "timeout_ceil_threshold": 5,
+                "max_line_size": 8190,
+                "max_field_size": 8190,
+                "max_headers": 128,
+            },
+            "timer": timer,
+            "timeout": timeout,
             "session": session,
             "ssl": True,
             "proxy_headers": None,