Add support for HTTP/2

CHANGES/13039.feature.rst

@@ -0,0 +1 @@
+Added experimental support for ``HTTP/2``. It can be enabled via the ``http2_enabled`` parameter of the :class:`~aiohttp.ClientSession` -- by :user:`Moist-Cat`.

aiohttp/client.py

@@ -98,6 +98,7 @@
     strip_auth_from_url,
 )
 from .http import WS_KEY, HttpVersion, WebSocketReader, WebSocketWriter
+from .http2.adapter import get_version
 from .http_websocket import WSHandshakeError, ws_ext_gen, ws_ext_parse
 from .tracing import Trace, TraceConfig
 from .typedefs import (
@@ -234,23 +235,74 @@ class _WSConnectOptions(TypedDict, total=False):
 async def _connect_and_send_request(req: ClientRequest) -> ClientResponse:
     connector = req._session._connector
     assert connector is not None
+    key = req.connection_key
     try:
+        # only the first connection to a host blocks
+        # the rest of the connection requests are done
+        # concurrently
+        await connector.semaphore.acquire(key)
+
         conn = await connector.connect(req, traces=req._traces, timeout=req._timeout)
+
+        connector.semaphore.release(key)
     except asyncio.TimeoutError as exc:
         raise ConnectionTimeoutError(f"Connection timeout to host {req.url}") from exc
+    finally:
+        connector.semaphore.release(key)
 
     assert conn.protocol is not None
-    conn.protocol.set_response_params(**req._response_params)
+    assert conn.protocol.transport is not None
+
+    alpn_protocol = get_version(conn.protocol)
+
+    resp = None
+    started = False
+
+    if alpn_protocol == "h2":
+        # release immediately to allow reuse
+        connector._release(conn._key, conn.protocol, should_close=False)
+        # the protocol corresponding to the connection
+        # remains (i.e., the count per host is always 1 for h2)
+        # This is the number of TCP connections not the number of
+        # streams
+        connector._acquired.add(conn.protocol)
     try:
-        resp = await req._send(conn)
-        try:
-            await resp.start(conn)
-        except BaseException:
+        # backwards compatibility
+        if alpn_protocol == "h2":
+            if req._continue:
+                # `expect100` is not inherently incompatible
+                # but the implementation does not appear to be
+                # straightforward.
+                # We have to set the result of the future in
+                # the HTTP/2 stream yet we can not close the stream
+                # because we have to wait for the second response.
+                #
+                # Also, it's not really necessary in HTTP/2
+                # because the server can simply reset the stream.
+                raise NotImplementedError("expect100 is not supported over HTTP/2")
+            stream = await conn.protocol.create_stream()  # type: ignore[attr-defined]
+            req.stream_id = stream.stream_id
+            # release again to clear the protocol from _acquired if required
+            connector._release(conn._key, conn.protocol, should_close=False)
+            conn.protocol.set_response_params(**req._response_params)
+            resp = await req._send(conn)
+            resp.stream_id = stream.stream_id
+        else:
+            conn.protocol.set_response_params(**req._response_params)
+            resp = await req._send(conn)
+        await resp.start(conn)
+
+        if alpn_protocol == "h2":
+            # we still have to null the protocol since we didn't close the connection
+            conn._protocol = None
+
+        started = True
+    finally:
+        if resp is not None and not started:
             resp.close()
-            raise
-    except BaseException:
-        conn.close()
-        raise
+            conn.close()
+        if resp is None:
+            conn.close()
     return resp
 
 
@@ -322,6 +374,7 @@ def __init__(
         fallback_charset_resolver: _CharsetResolver = lambda r, b: "utf-8",
         middlewares: Sequence[ClientMiddlewareType] = (),
         ssl_shutdown_timeout: _SENTINEL | None | float = sentinel,
+        http2_enabled: bool = False,
     ) -> None:
         # We initialise _connector to None immediately, as it's referenced in __del__()
         # and could cause issues if an exception occurs during initialisation.
@@ -361,7 +414,9 @@ def __init__(
             )
 
         if connector is None:
-            connector = TCPConnector(ssl_shutdown_timeout=ssl_shutdown_timeout)
+            connector = TCPConnector(
+                ssl_shutdown_timeout=ssl_shutdown_timeout, http2_enabled=http2_enabled
+            )
         # Initialize these three attrs before raising any exception,
         # they are used in __del__
         self._connector = connector
@@ -1594,7 +1649,8 @@ def request(
         connector_owner = False
         if connector is None:
             connector_owner = True
-            connector = TCPConnector(force_close=True)
+            http2_enabled = kwargs.get("http2_enabled", False)
+            connector = TCPConnector(force_close=True, http2_enabled=http2_enabled)
 
         session = ClientSession(
             cookies=kwargs.pop("cookies", None),

aiohttp/client_reqrep.py

@@ -59,6 +59,7 @@
     HttpVersion11,
     StreamWriter,
 )
+from .http2.adapter import Http2StreamWriter, get_version
 from .streams import EMPTY_PAYLOAD, StreamReader
 from .typedefs import DEFAULT_JSON_DECODER, JSONDecoder, RawHeaders
 
@@ -280,6 +281,9 @@ class ClientResponse(HeadersMixin):
     _output_size: int = 0
     _upload_complete: asyncio.Future[None] | None = None
 
+    # HTTP/2 stream id
+    stream_id: int | None = None
+
     def __init__(
         self,
         method: str,
@@ -525,7 +529,13 @@ async def start(self, connection: "Connection") -> "ClientResponse":
                 # read response
                 try:
                     protocol = self._protocol
-                    message, payload = await protocol.read()  # type: ignore[union-attr]
+                    # conditional branching to pass the stream id
+                    # to the protocol
+                    assert protocol is not None
+                    if get_version(protocol) == "h2":
+                        message, payload = await protocol.read_stream(self.stream_id)  # type: ignore[attr-defined]
+                    else:
+                        message, payload = await protocol.read()
                 except HttpProcessingError as exc:
                     raise ClientResponseError(
                         self.request_info,
@@ -810,6 +820,9 @@ class ClientRequestBase:
 
     _skip_auto_headers: "CIMultiDict[None] | None" = None
 
+    # HTTP/2 stream id
+    stream_id: int | None = None
+
     # N.B.
     # Adding __del__ method with self._writer closing doesn't make sense
     # because _writer is instance method, thus it keeps a reference to self.
@@ -932,7 +945,9 @@ def _create_response(
             stream_writer=stream_writer,
         )
 
-    def _create_writer(self, protocol: BaseProtocol) -> StreamWriter:
+    def _create_writer(
+        self, protocol: BaseProtocol
+    ) -> StreamWriter | Http2StreamWriter:
         return StreamWriter(protocol, self.loop)
 
     def _should_write(self, protocol: BaseProtocol) -> bool:
@@ -1428,7 +1443,13 @@ def _create_response(
             stream_writer=stream_writer,
         )
 
-    def _create_writer(self, protocol: BaseProtocol) -> StreamWriter:
+    def _create_writer(
+        self, protocol: BaseProtocol
+    ) -> StreamWriter | Http2StreamWriter:
+        if get_version(protocol) == "h2":
+            if self.compress:
+                raise ValueError("Payload compression is not supported over HTTP/2")
+            return Http2StreamWriter(protocol, self.loop, self)
         writer = StreamWriter(
             protocol,
             self.loop,

aiohttp/connector.py

@@ -10,7 +10,7 @@
 from collections.abc import Awaitable, Callable, Iterator, Sequence
 from contextlib import suppress
 from http import HTTPStatus
-from itertools import chain, cycle, islice
+from itertools import chain, cycle, islice, product
 from time import monotonic
 from types import TracebackType
 from typing import TYPE_CHECKING, Any, Literal, cast
@@ -52,6 +52,8 @@
     set_exception,
     set_result,
 )
+from .http2.synchro import HostProbeSynchronizer
+from .http_protocol import HttpDispatcherProtocol
 from .log import client_logger
 from .resolver import DefaultResolver
 
@@ -109,7 +111,7 @@ async def create_connection(
     ssl_shutdown_timeout: float | None = None,
 ) -> tuple[asyncio.Transport, ResponseHandler]:
     if aiofastnet is not None:
-        return await aiofastnet.create_connection(
+        transport, proto = await aiofastnet.create_connection(
             loop,
             protocol_factory,
             ssl=ssl,
@@ -119,26 +121,28 @@ async def create_connection(
         )
     else:
         if sys.version_info >= (3, 11):  # type: ignore[unreachable]
-            return await loop.create_connection(
+            transport, proto = await loop.create_connection(
                 protocol_factory,
                 ssl=ssl,
                 sock=sock,
                 server_hostname=server_hostname,
                 ssl_shutdown_timeout=ssl_shutdown_timeout,
             )
         else:
-            return await loop.create_connection(
+            transport, proto = await loop.create_connection(
                 protocol_factory,
                 ssl=ssl,
                 sock=sock,
                 server_hostname=server_hostname,
             )
 
+    return transport, proto._handler  # type: ignore[attr-defined]
+
 
 async def start_tls(
     loop: asyncio.AbstractEventLoop,
     transport: asyncio.Transport,
-    protocol: ResponseHandler,
+    protocol: HttpDispatcherProtocol | ResponseHandler,
     sslcontext: SSLContext,
     *,
     server_hostname: str | None,
@@ -375,7 +379,7 @@ def __init__(
         ] = defaultdict(OrderedDict)
 
         self._loop = loop
-        self._factory = functools.partial(ResponseHandler, loop=loop)
+        self._factory = functools.partial(HttpDispatcherProtocol, loop=loop)
 
         # start keep-alive connection cleanup task
         self._cleanup_handle: asyncio.TimerHandle | None = None
@@ -402,6 +406,12 @@ def __init__(
         self._placeholder_future.set_result(None)
         self._cleanup_closed()
 
+        # Semaphore for HTTP/2 connections
+        # avoids duplicate connections to the
+        # same host
+        # (HTTP/2 doesn't need connection pooling to send multiple requests)
+        self.semaphore = HostProbeSynchronizer()
+
     def __del__(self, _warnings: Any = warnings) -> None:
         if self._closed:
             return
@@ -920,7 +930,7 @@ def expired(self, key: tuple[str, int]) -> bool:
         return self._timestamps[key] + self._ttl < monotonic()
 
 
-def _make_ssl_context(verified: bool) -> SSLContext:
+def _make_ssl_context(verified: bool, http2_enabled: bool = False) -> SSLContext:
     """Create SSL context.
 
     This method is not async-friendly and should be called from a thread
@@ -939,16 +949,28 @@ def _make_ssl_context(verified: bool) -> SSLContext:
         sslcontext.verify_mode = ssl.CERT_NONE
         sslcontext.options |= ssl.OP_NO_COMPRESSION
         sslcontext.set_default_verify_paths()
-    sslcontext.set_alpn_protocols(("http/1.1",))
+
+    protocols = ["http/1.1"]
+    if http2_enabled:
+        protocols += ["h2"]
+    sslcontext.set_alpn_protocols(tuple(protocols))
     return sslcontext
 
 
+# map configurations to ssl context
+# enable_http2, enable_http3, enable_http4, ...
+OPTIONAL_PROTOCOLS = 1
+_SSL_CONTEXT_MAP = {}
+for verified in (True, False):
+    for mask in product((True, False), repeat=OPTIONAL_PROTOCOLS):
+        _SSL_CONTEXT_MAP[(verified,) + mask] = _make_ssl_context(verified, *mask)
+
 # The default SSLContext objects are created at import time
 # since they do blocking I/O to load certificates from disk,
 # and imports should always be done before the event loop starts
 # or in a thread.
-_SSL_CONTEXT_VERIFIED = _make_ssl_context(True)
-_SSL_CONTEXT_UNVERIFIED = _make_ssl_context(False)
+_SSL_CONTEXT_VERIFIED = _SSL_CONTEXT_MAP[(True, False)]
+_SSL_CONTEXT_UNVERIFIED = _SSL_CONTEXT_MAP[(False, False)]
 
 
 class TCPConnector(BaseConnector):
@@ -1012,6 +1034,7 @@ def __init__(
         interleave: int | None = None,
         socket_factory: SocketFactoryType | None = None,
         ssl_shutdown_timeout: _SENTINEL | None | float = sentinel,
+        http2_enabled: bool = False,
     ):
         super().__init__(
             keepalive_timeout=keepalive_timeout,
@@ -1051,6 +1074,7 @@ def __init__(
         self._resolve_host_tasks: set[asyncio.Task[list[ResolveResult]]] = set()
         self._socket_factory = socket_factory
         self._ssl_shutdown_timeout: float | None
+        self._http2_enabled = http2_enabled
 
         # Handle ssl_shutdown_timeout with warning for Python < 3.11
         if ssl_shutdown_timeout is sentinel:
@@ -1304,14 +1328,14 @@ def _get_ssl_context(self, req: ClientRequestBase) -> SSLContext | None:
             return sslcontext
         if sslcontext is not True:
             # not verified or fingerprinted
-            return _SSL_CONTEXT_UNVERIFIED
+            return _SSL_CONTEXT_MAP[(False, self._http2_enabled)]
         sslcontext = self._ssl
         if isinstance(sslcontext, ssl.SSLContext):
             return sslcontext
         if sslcontext is not True:
             # not verified or fingerprinted
-            return _SSL_CONTEXT_UNVERIFIED
-        return _SSL_CONTEXT_VERIFIED
+            return _SSL_CONTEXT_MAP[(False, self._http2_enabled)]
+        return _SSL_CONTEXT_MAP[(True, self._http2_enabled)]
 
     def _get_fingerprint(self, req: ClientRequestBase) -> "Fingerprint | None":
         ret = req.ssl
@@ -1499,7 +1523,9 @@ async def _start_tls_connection(
                 tls_transport
             )  # Kick the state machine of the new TLS protocol
 
-        return tls_transport, tls_proto
+        # HACK use the correct type
+        proto = tls_proto._handler
+        return tls_transport, proto  # type: ignore[return-value]
 
     def _convert_hosts_to_addr_infos(
         self, hosts: list[ResolveResult]
@@ -1591,7 +1617,6 @@ async def _create_direct_connection(
                     bad_peer = sock.getpeername()
                     aiohappyeyeballs.remove_addr_infos(addr_infos, bad_peer)
                     continue
-
             return transp, proto
         assert last_exc is not None
         raise last_exc
@@ -1723,7 +1748,7 @@ async def _create_connection(
                 raise
             raise UnixClientConnectorError(self.path, req.connection_key, exc) from exc
 
-        return proto
+        return proto  # type: ignore[return-value]
 
 
 class NamedPipeConnector(BaseConnector):

aiohttp/helpers.py

@@ -47,7 +47,6 @@
 
 from . import hdrs
 from .log import client_logger
-from .typedefs import PathLike  # noqa
 
 if sys.version_info >= (3, 11):
     import asyncio as async_timeout