Eliminate multi-assignment churn in the connection pool

src/httpcore2/httpcore2/_async/connection.py

@@ -177,6 +177,9 @@ def is_idle(self) -> bool:
             return self._connect_failed
         return self._connection.is_idle()
 
+    def can_multiplex(self) -> bool:
+        return self._connection is not None and self._connection.can_multiplex()
+
     def is_closed(self) -> bool:
         if self._connection is None:
             return self._connect_failed

src/httpcore2/httpcore2/_async/connection_pool.py

@@ -264,34 +264,40 @@ def _assign_requests_to_connections(self) -> list[AsyncConnectionInterface]:
         closing_connections: list[AsyncConnectionInterface] = []
         retained_connections: list[AsyncConnectionInterface] = []
 
-        # Connections currently referenced by an active request (including
-        # connections that are in the process of being established).
+        # Connections currently referenced by an in-flight request, including
+        # connections that are in the process of being established and idle
+        # connections reserved by an assigned-but-not-yet-sent request.
         request_connections = {r.connection for r in self._requests}
 
         # First we handle cleaning up any connections that are closed
-        # or have expired their keep-alive, in a single pass.
+        # or have expired their keep-alive, in a single pass. Reserved
+        # connections skip the expiry check: they were checked when assigned,
+        # and `has_expired()` on an idle connection probes the socket.
         for connection in self._connections:
+            reserved = connection in request_connections
             if connection.is_closed():
                 continue
-            elif not (connection.is_connected() or connection in request_connections):
+            elif not (connection.is_connected() or reserved):
                 # Garbage: a NEW-state connection whose request was cancelled
                 # before the TCP handshake completed.  Drop it without closing
                 # (there is no socket to close yet).
                 continue
-            elif connection.has_expired():
+            elif not reserved and connection.has_expired():
                 closing_connections.append(connection)
             else:
                 retained_connections.append(connection)
 
         # Then we close any surplus idle connections, to enforce the
-        # max_keepalive_connections setting.
+        # max_keepalive_connections setting. Reserved connections are not
+        # surplus: a request is about to be sent on them.
         idle_surplus = (
-            sum(connection.is_idle() for connection in retained_connections) - self._max_keepalive_connections
+            sum(connection.is_idle() and connection not in request_connections for connection in retained_connections)
+            - self._max_keepalive_connections
         )
         if idle_surplus > 0:
             kept: list[AsyncConnectionInterface] = []
             for connection in retained_connections:
-                if idle_surplus > 0 and connection.is_idle():
+                if idle_surplus > 0 and connection.is_idle() and connection not in request_connections:
                     closing_connections.append(connection)
                     idle_surplus -= 1
                 else:
@@ -303,11 +309,27 @@ def _assign_requests_to_connections(self) -> list[AsyncConnectionInterface]:
         # Snapshot the set of reusable connections once, rather than rebuilding
         # it per queued request — this is what brings the loop from O(N*M) to
         # O(N+M) in the common case.
-        available_connections = [connection for connection in self._connections if connection.is_available()]
+        #
+        # An idle connection already assigned to an in-flight request is
+        # reserved: it stays IDLE until the winning task sends on it, so
+        # without this exclusion the next pass would assign it again and the
+        # loser would churn through `ConnectionNotAvailable`. Multiplexing
+        # connections are exempt: they can take further requests while idle.
+        available_connections = [
+            connection
+            for connection in self._connections
+            if connection.is_available()
+            and not (connection.is_idle() and connection in request_connections and not connection.can_multiplex())
+        ]
         new_connection_budget = self._max_connections - len(self._connections)
 
-        # Assign queued requests to connections.
+        # Assign queued requests to connections. Once no connection is
+        # available and no new connection may be created, no queued request
+        # can be assigned, so the scan stops early: this keeps a pass on a
+        # saturated pool O(connections) rather than O(in-flight requests).
         for pool_request in self._requests:
+            if not available_connections and new_connection_budget <= 0:
+                break
             if not pool_request.is_queued():
                 continue
             origin = pool_request.request.url.origin
@@ -318,9 +340,13 @@ def _assign_requests_to_connections(self) -> list[AsyncConnectionInterface]:
             # 2. We can create a new connection to handle the request.
             # 3. We can close an idle connection and then create a new connection
             #    to handle the request.
-            for connection in available_connections:
+            for idx, connection in enumerate(available_connections):
                 if connection.can_handle_request(origin):
                     pool_request.assign_to_connection(connection)
+                    if connection.is_idle() and not connection.can_multiplex():
+                        # An idle HTTP/1.1 connection can only take this
+                        # single request until it is released.
+                        del available_connections[idx]
                     break
             else:
                 if new_connection_budget > 0:

src/httpcore2/httpcore2/_async/http2.py

@@ -495,6 +495,9 @@ def has_expired(self) -> bool:
     def is_idle(self) -> bool:
         return self._state == HTTPConnectionState.IDLE
 
+    def can_multiplex(self) -> bool:
+        return True
+
     def is_closed(self) -> bool:
         return self._state == HTTPConnectionState.CLOSED
 

src/httpcore2/httpcore2/_async/interfaces.py

@@ -140,6 +140,16 @@ def is_idle(self) -> bool:
         """
         raise NotImplementedError()  # pragma: no cover
 
+    def can_multiplex(self) -> bool:
+        """
+        Return `True` if the connection can serve multiple requests
+        concurrently, such as an established HTTP/2 connection.
+
+        The default covers HTTP/1.1-style implementations, which serve a
+        single request at a time.
+        """
+        return False
+
     def is_closed(self) -> bool:
         """
         Return `True` if the connection has been closed.

src/httpcore2/httpcore2/_sync/connection.py

@@ -177,6 +177,9 @@ def is_idle(self) -> bool:
             return self._connect_failed
         return self._connection.is_idle()
 
+    def can_multiplex(self) -> bool:
+        return self._connection is not None and self._connection.can_multiplex()
+
     def is_closed(self) -> bool:
         if self._connection is None:
             return self._connect_failed

src/httpcore2/httpcore2/_sync/connection_pool.py

@@ -264,34 +264,40 @@ def _assign_requests_to_connections(self) -> list[ConnectionInterface]:
         closing_connections: list[ConnectionInterface] = []
         retained_connections: list[ConnectionInterface] = []
 
-        # Connections currently referenced by an active request (including
-        # connections that are in the process of being established).
+        # Connections currently referenced by an in-flight request, including
+        # connections that are in the process of being established and idle
+        # connections reserved by an assigned-but-not-yet-sent request.
         request_connections = {r.connection for r in self._requests}
 
         # First we handle cleaning up any connections that are closed
-        # or have expired their keep-alive, in a single pass.
+        # or have expired their keep-alive, in a single pass. Reserved
+        # connections skip the expiry check: they were checked when assigned,
+        # and `has_expired()` on an idle connection probes the socket.
         for connection in self._connections:
+            reserved = connection in request_connections
             if connection.is_closed():
                 continue
-            elif not (connection.is_connected() or connection in request_connections):
+            elif not (connection.is_connected() or reserved):
                 # Garbage: a NEW-state connection whose request was cancelled
                 # before the TCP handshake completed.  Drop it without closing
                 # (there is no socket to close yet).
                 continue
-            elif connection.has_expired():
+            elif not reserved and connection.has_expired():
                 closing_connections.append(connection)
             else:
                 retained_connections.append(connection)
 
         # Then we close any surplus idle connections, to enforce the
-        # max_keepalive_connections setting.
+        # max_keepalive_connections setting. Reserved connections are not
+        # surplus: a request is about to be sent on them.
         idle_surplus = (
-            sum(connection.is_idle() for connection in retained_connections) - self._max_keepalive_connections
+            sum(connection.is_idle() and connection not in request_connections for connection in retained_connections)
+            - self._max_keepalive_connections
         )
         if idle_surplus > 0:
             kept: list[ConnectionInterface] = []
             for connection in retained_connections:
-                if idle_surplus > 0 and connection.is_idle():
+                if idle_surplus > 0 and connection.is_idle() and connection not in request_connections:
                     closing_connections.append(connection)
                     idle_surplus -= 1
                 else:
@@ -303,11 +309,27 @@ def _assign_requests_to_connections(self) -> list[ConnectionInterface]:
         # Snapshot the set of reusable connections once, rather than rebuilding
         # it per queued request — this is what brings the loop from O(N*M) to
         # O(N+M) in the common case.
-        available_connections = [connection for connection in self._connections if connection.is_available()]
+        #
+        # An idle connection already assigned to an in-flight request is
+        # reserved: it stays IDLE until the winning task sends on it, so
+        # without this exclusion the next pass would assign it again and the
+        # loser would churn through `ConnectionNotAvailable`. Multiplexing
+        # connections are exempt: they can take further requests while idle.
+        available_connections = [
+            connection
+            for connection in self._connections
+            if connection.is_available()
+            and not (connection.is_idle() and connection in request_connections and not connection.can_multiplex())
+        ]
         new_connection_budget = self._max_connections - len(self._connections)
 
-        # Assign queued requests to connections.
+        # Assign queued requests to connections. Once no connection is
+        # available and no new connection may be created, no queued request
+        # can be assigned, so the scan stops early: this keeps a pass on a
+        # saturated pool O(connections) rather than O(in-flight requests).
         for pool_request in self._requests:
+            if not available_connections and new_connection_budget <= 0:
+                break
             if not pool_request.is_queued():
                 continue
             origin = pool_request.request.url.origin
@@ -318,9 +340,13 @@ def _assign_requests_to_connections(self) -> list[ConnectionInterface]:
             # 2. We can create a new connection to handle the request.
             # 3. We can close an idle connection and then create a new connection
             #    to handle the request.
-            for connection in available_connections:
+            for idx, connection in enumerate(available_connections):
                 if connection.can_handle_request(origin):
                     pool_request.assign_to_connection(connection)
+                    if connection.is_idle() and not connection.can_multiplex():
+                        # An idle HTTP/1.1 connection can only take this
+                        # single request until it is released.
+                        del available_connections[idx]
                     break
             else:
                 if new_connection_budget > 0:

src/httpcore2/httpcore2/_sync/http2.py

@@ -495,6 +495,9 @@ def has_expired(self) -> bool:
     def is_idle(self) -> bool:
         return self._state == HTTPConnectionState.IDLE
 
+    def can_multiplex(self) -> bool:
+        return True
+
     def is_closed(self) -> bool:
         return self._state == HTTPConnectionState.CLOSED
 

src/httpcore2/httpcore2/_sync/interfaces.py

@@ -140,6 +140,16 @@ def is_idle(self) -> bool:
         """
         raise NotImplementedError()  # pragma: no cover
 
+    def can_multiplex(self) -> bool:
+        """
+        Return `True` if the connection can serve multiple requests
+        concurrently, such as an established HTTP/2 connection.
+
+        The default covers HTTP/1.1-style implementations, which serve a
+        single request at a time.
+        """
+        return False
+
     def is_closed(self) -> bool:
         """
         Return `True` if the connection has been closed.

tests/httpcore2/_async/test_connection_pool.py

@@ -789,3 +789,98 @@ async def trace(name: str, kwargs: dict[str, typing.Any]) -> None:
         "http11.response_closed.started",
         "http11.response_closed.complete",
     ]
+
+
+@pytest.mark.trio
+async def test_connection_pool_assigns_released_connection_to_one_queued_request() -> None:
+    """
+    A released connection must be handed to exactly one queued request.
+
+    Assigning it to every queued request wakes them all, only for all but one
+    to fail with `ConnectionNotAvailable` and re-enter the queue, degrading
+    quadratically with queue depth.
+    """
+
+    class CountingPool(httpcore2.AsyncConnectionPool):
+        assign_passes = 0
+
+        def _assign_requests_to_connections(self) -> list[httpcore2.AsyncConnectionInterface]:
+            CountingPool.assign_passes += 1
+            return super()._assign_requests_to_connections()
+
+    network_backend = httpcore2.AsyncMockBackend(
+        [
+            b"HTTP/1.1 200 OK\r\n",
+            b"Content-Type: plain/text\r\n",
+            b"Content-Length: 13\r\n",
+            b"\r\n",
+            b"Hello, world!",
+        ]
+        * 10
+    )
+
+    async def fetch(pool: httpcore2.AsyncConnectionPool) -> None:
+        async with pool.stream("GET", "https://example.com/") as response:
+            await response.aread()
+        assert response.status == 200
+
+    async with CountingPool(max_connections=1, network_backend=network_backend) as pool:
+        async with concurrency.open_nursery() as nursery:
+            for _ in range(10):
+                nursery.start_soon(fetch, pool)
+
+    # Exactly two passes per request: one when it is queued, one when it
+    # releases its connection.
+    assert CountingPool.assign_passes == 2 * 10
+
+
+@pytest.mark.trio
+async def test_connection_pool_multiplexes_idle_http2_connection_within_a_pass() -> None:
+    """
+    A burst of requests arriving while a warmed HTTP/2 connection is idle
+    must be assigned to it immediately, not serialized behind the first
+    request's reservation.
+    """
+
+    class QueueObservingPool(httpcore2.AsyncConnectionPool):
+        max_queued_after_pass = 0
+
+        def _assign_requests_to_connections(self) -> list[httpcore2.AsyncConnectionInterface]:
+            closing = super()._assign_requests_to_connections()
+            queued = sum(request.is_queued() for request in self._requests)
+            QueueObservingPool.max_queued_after_pass = max(QueueObservingPool.max_queued_after_pass, queued)
+            return closing
+
+    def response_frames(stream_id: int) -> list[bytes]:
+        return [
+            hyperframe.frame.HeadersFrame(
+                stream_id=stream_id,
+                data=hpack.Encoder().encode([(b":status", b"200")]),
+                flags=["END_HEADERS"],
+            ).serialize(),
+            hyperframe.frame.DataFrame(stream_id=stream_id, data=b"Hello, world!", flags=["END_STREAM"]).serialize(),
+        ]
+
+    network_backend = httpcore2.AsyncMockBackend(
+        buffer=[
+            hyperframe.frame.SettingsFrame().serialize(),
+            *response_frames(1),
+            *response_frames(3),
+            *response_frames(5),
+            *response_frames(7),
+        ],
+        http2=True,
+    )
+
+    async def fetch(pool: httpcore2.AsyncConnectionPool) -> None:
+        response = await pool.request("GET", "https://example.com/")
+        assert response.status == 200
+
+    async with QueueObservingPool(network_backend=network_backend, max_connections=1, http2=True) as pool:
+        # Warm the connection; it returns to the pool IDLE.
+        await fetch(pool)
+        async with concurrency.open_nursery() as nursery:
+            for _ in range(3):
+                nursery.start_soon(fetch, pool)
+
+    assert QueueObservingPool.max_queued_after_pass == 0