Implement happy eyeballs (RFC 8305)

aiohttp/connector.py

@@ -3,6 +3,7 @@
 import functools
 import logging
 import random
+import socket
 import sys
 import traceback
 import warnings
@@ -31,6 +32,8 @@
     cast,
 )
 
+import aiohappyeyeballs
+
 from . import hdrs, helpers
 from .abc import AbstractResolver
 from .client_exceptions import (
@@ -69,6 +72,39 @@
     from .tracing import Trace
 
 
+def _convert_hosts_to_addr_infos(
+    hosts: List[Dict[str, Any]]
+) -> List[aiohappyeyeballs.AddrInfoType]:
+    """Converts the list of hosts to a list of addr_infos.
+
+    The list of hosts is the result of a DNS lookup. The list of
+    addr_infos is the result of a call to `socket.getaddrinfo()`.
+    """
+    addr_infos: List[aiohappyeyeballs.AddrInfoType] = []
+    for hinfo in hosts:
+        host = hinfo["host"]
+        is_ipv6 = helpers.is_ipv6_address(host)
+        family = socket.AF_INET6 if is_ipv6 else socket.AF_INET
+        addr = (host, hinfo["port"], 0, 0) if is_ipv6 else (host, hinfo["port"])
+        addr_infos.append(
+            (family, socket.SOCK_STREAM, socket.IPPROTO_TCP, hinfo["hostname"], addr)
+        )
+    return addr_infos
+
+
+def _convert_local_addr_to_addr_infos(
+    local_addr: Optional[Tuple[str, int]]
+) -> Optional[List[aiohappyeyeballs.AddrInfoType]]:
+    """Convert a local_addr tuple to a list of addr_infos."""
+    if local_addr is None:
+        return None
+    host, port = local_addr
+    is_ipv6 = helpers.is_ipv6_address(host)
+    family = socket.AF_INET6 if is_ipv6 else socket.AF_INET
+    addr = (host, port, 0, 0) if is_ipv6 else (host, port)
+    return [(family, socket.SOCK_STREAM, socket.IPPROTO_TCP, "", addr)]
+
+
 class Connection:
     _source_traceback = None
     _transport = None
@@ -748,6 +784,8 @@ def __init__(
         limit_per_host: int = 0,
         enable_cleanup_closed: bool = False,
         timeout_ceil_threshold: float = 5,
+        happy_eyeballs_delay: float = 0.25,
+        interleave: int = 1,
     ) -> None:
         super().__init__(
             keepalive_timeout=keepalive_timeout,
@@ -772,7 +810,9 @@ def __init__(
         self._cached_hosts = _DNSCacheTable(ttl=ttl_dns_cache)
         self._throttle_dns_events: Dict[Tuple[str, int], EventResultOrError] = {}
         self._family = family
-        self._local_addr = local_addr
+        self._local_addr_infos = _convert_local_addr_to_addr_infos(local_addr)
+        self._happy_eyeballs_delay = happy_eyeballs_delay
+        self._interleave = interleave
 
     def _close_immediately(self) -> List["asyncio.Future[None]"]:
         for ev in self._throttle_dns_events.values():
@@ -956,6 +996,7 @@ def _get_fingerprint(self, req: ClientRequest) -> Optional["Fingerprint"]:
     async def _wrap_create_connection(
         self,
         *args: Any,
+        addr_infos: List[aiohappyeyeballs.AddrInfoType],
         req: ClientRequest,
         timeout: "ClientTimeout",
         client_error: Type[Exception] = ClientConnectorError,
@@ -965,7 +1006,14 @@ async def _wrap_create_connection(
             async with ceil_timeout(
                 timeout.sock_connect, ceil_threshold=timeout.ceil_threshold
             ):
-                return await self._loop.create_connection(*args, **kwargs)
+                sock = await aiohappyeyeballs.start_connection(
+                    addr_infos=addr_infos,
+                    local_addr_infos=self._local_addr_infos,
+                    happy_eyeballs_delay=self._happy_eyeballs_delay,
+                    interleave=self._interleave,
+                    loop=self._loop,
+                )
+                return await self._loop.create_connection(*args, **kwargs, sock=sock)
         except cert_errors as exc:
             raise ClientConnectorCertificateError(req.connection_key, exc) from exc
         except ssl_errors as exc:
@@ -1120,36 +1168,27 @@ def drop_exception(fut: "asyncio.Future[List[Dict[str, Any]]]") -> None:
             raise ClientConnectorError(req.connection_key, exc) from exc
 
         last_exc: Optional[Exception] = None
-
-        for hinfo in hosts:
-            host = hinfo["host"]
-            port = hinfo["port"]
-
+        addr_infos = _convert_hosts_to_addr_infos(hosts)
+        while addr_infos:
             # Strip trailing dots, certificates contain FQDN without dots.
             # See https://github.com/aio-libs/aiohttp/issues/3636
             server_hostname = (
-                (req.server_hostname or hinfo["hostname"]).rstrip(".")
-                if sslcontext
-                else None
+                (req.server_hostname or host).rstrip(".") if sslcontext else None
             )
 
             try:
                 transp, proto = await self._wrap_create_connection(
                     self._factory,
-                    host,
-                    port,
                     timeout=timeout,
                     ssl=sslcontext,
-                    family=hinfo["family"],
-                    proto=hinfo["proto"],
-                    flags=hinfo["flags"],
+                    addr_infos=addr_infos,
                     server_hostname=server_hostname,
-                    local_addr=self._local_addr,
                     req=req,
                     client_error=client_error,
                 )
             except ClientConnectorError as exc:
                 last_exc = exc
+                aiohappyeyeballs.pop_addr_infos_interleave(addr_infos, self._interleave)
                 continue
 
             if req.is_ssl() and fingerprint:
@@ -1160,6 +1199,10 @@ def drop_exception(fut: "asyncio.Future[List[Dict[str, Any]]]") -> None:
                     if not self._cleanup_closed_disabled:
                         self._cleanup_closed_transports.append(transp)
                     last_exc = exc
+                    # Remove the bad peer from the list of addr_infos
+                    sock: socket.socket = transp.get_extra_info("socket")
+                    bad_peer = sock.getpeername()
+                    aiohappyeyeballs.remove_addr_infos(addr_infos, bad_peer[0])
                     continue
 
             return transp, proto

requirements/base.txt

@@ -6,6 +6,8 @@
 #
 aiodns==3.1.1 ; sys_platform == "linux" or sys_platform == "darwin"
     # via -r requirements/runtime-deps.in
+aiohappyeyeballs==1.8.1
+    # via -r requirements/runtime-deps.in
 aiosignal==1.3.1
     # via -r requirements/runtime-deps.in
 async-timeout==4.0.3 ; python_version < "3.11"

requirements/constraints.txt

@@ -1,11 +1,13 @@
 #
-# This file is autogenerated by pip-compile with python 3.8
-# To update, run:
+# This file is autogenerated by pip-compile with Python 3.8
+# by the following command:
 #
-#    pip-compile --allow-unsafe --output-file=requirements/constraints.txt --resolver=backtracking --strip-extras requirements/constraints.in
+#    pip-compile --allow-unsafe --output-file=requirements/constraints.txt --strip-extras requirements/constraints.in
 #
 aiodns==3.1.1 ; sys_platform == "linux" or sys_platform == "darwin"
     # via -r requirements/runtime-deps.in
+aiohappyeyeballs==1.8.1
+    # via -r requirements/runtime-deps.in
 aiohttp-theme==0.1.6
     # via -r requirements/doc.in
 aioredis==2.0.1

requirements/dev.txt

@@ -1,11 +1,13 @@
 #
-# This file is autogenerated by pip-compile with python 3.8
-# To update, run:
+# This file is autogenerated by pip-compile with Python 3.8
+# by the following command:
 #
-#    pip-compile --allow-unsafe --output-file=requirements/dev.txt --resolver=backtracking --strip-extras requirements/dev.in
+#    pip-compile --allow-unsafe --output-file=requirements/dev.txt --strip-extras requirements/dev.in
 #
 aiodns==3.1.1 ; sys_platform == "linux" or sys_platform == "darwin"
     # via -r requirements/runtime-deps.in
+aiohappyeyeballs==1.8.1
+    # via -r requirements/runtime-deps.in
 aiohttp-theme==0.1.6
     # via -r requirements/doc.in
 aioredis==2.0.1

requirements/runtime-deps.in

@@ -1,6 +1,7 @@
 # Extracted from `setup.cfg` via `make sync-direct-runtime-deps`
 
 aiodns >= 1.1; sys_platform=="linux" or sys_platform=="darwin"
+aiohappyeyeballs >= 1.8.1
 aiosignal >= 1.1.2
 async-timeout >= 4.0, < 5.0 ; python_version < "3.11"
 Brotli; platform_python_implementation == 'CPython'

requirements/runtime-deps.txt

@@ -6,6 +6,8 @@
 #
 aiodns==3.1.1 ; sys_platform == "linux" or sys_platform == "darwin"
     # via -r requirements/runtime-deps.in
+aiohappyeyeballs==1.8.1
+    # via -r requirements/runtime-deps.in
 aiosignal==1.3.1
     # via -r requirements/runtime-deps.in
 async-timeout==4.0.3 ; python_version < "3.11"

requirements/test.txt

@@ -1,11 +1,13 @@
 #
-# This file is autogenerated by pip-compile with python 3.8
+# This file is autogenerated by pip-compile with Python 3.8
 # by the following command:
 #
-#    pip-compile --allow-unsafe --output-file=requirements/test.txt --resolver=backtracking --strip-extras requirements/test.in
+#    pip-compile --allow-unsafe --output-file=requirements/test.txt --strip-extras requirements/test.in
 #
 aiodns==3.1.1 ; sys_platform == "linux" or sys_platform == "darwin"
     # via -r requirements/runtime-deps.in
+aiohappyeyeballs==1.8.1
+    # via -r requirements/runtime-deps.in
 aiosignal==1.3.1
     # via -r requirements/runtime-deps.in
 async-timeout==4.0.3 ; python_version < "3.11"

setup.cfg

@@ -47,6 +47,7 @@ zip_safe = False
 include_package_data = True
 
 install_requires =
+  aiohappyeyeballs >= 1.8.1
   aiosignal >= 1.1.2
   async-timeout >= 4.0, < 5.0 ; python_version < "3.11"
   frozenlist >= 1.1.1

tests/conftest.py

@@ -208,3 +208,11 @@ def netrc_contents(
     monkeypatch.setenv("NETRC", str(netrc_file_path))
 
     return netrc_file_path
+
+
+@pytest.fixture
+def start_connection():
+    with mock.patch(
+        "aiohttp.connector.aiohappyeyeballs.start_connection"
+    ) as start_connection_mock:
+        yield start_connection_mock