Add support for HTTP/2

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,62 @@ 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":
+            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)
+            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
 
 

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,11 @@ 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":
+            return Http2StreamWriter(protocol, self.loop, self)
         writer = StreamWriter(
             protocol,
             self.loop,

aiohttp/connector.py

@@ -1,5 +1,6 @@
 import asyncio
 import functools
+import os
 import random
 import socket
 import sys
@@ -52,6 +53,8 @@
     set_exception,
     set_result,
 )
+from .http2.synchro import HostProbeSynchronizer
+from .http_protocol import HttpDispatcherProtocol
 from .log import client_logger
 from .resolver import DefaultResolver
 
@@ -138,7 +141,7 @@ async def create_connection(
 async def start_tls(
     loop: asyncio.AbstractEventLoop,
     transport: asyncio.Transport,
-    protocol: ResponseHandler,
+    protocol: HttpDispatcherProtocol | ResponseHandler,
     sslcontext: SSLContext,
     *,
     server_hostname: str | None,
@@ -375,7 +378,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 +405,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
@@ -939,7 +948,11 @@ 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 os.getenv("AIOHTTP_ENABLE_EXPERIMENTAL_PROTOCOLS", False):
+        protocols += ["h2"]
+    sslcontext.set_alpn_protocols(tuple(protocols))
     return sslcontext
 
 
@@ -1499,7 +1512,8 @@ 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
+        return tls_transport, tls_proto  # type: ignore[return-value]
 
     def _convert_hosts_to_addr_infos(
         self, hosts: list[ResolveResult]
@@ -1591,7 +1605,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 +1736,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