diff --git a/src/httpcore2/httpcore2/_async/connection.py b/src/httpcore2/httpcore2/_async/connection.py index ff3509bd..e3fb8596 100644 --- a/src/httpcore2/httpcore2/_async/connection.py +++ b/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 diff --git a/src/httpcore2/httpcore2/_async/connection_pool.py b/src/httpcore2/httpcore2/_async/connection_pool.py index 7011176e..2c5e3980 100644 --- a/src/httpcore2/httpcore2/_async/connection_pool.py +++ b/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: diff --git a/src/httpcore2/httpcore2/_async/http2.py b/src/httpcore2/httpcore2/_async/http2.py index 6f2b4b35..7d2dab5d 100644 --- a/src/httpcore2/httpcore2/_async/http2.py +++ b/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 diff --git a/src/httpcore2/httpcore2/_async/interfaces.py b/src/httpcore2/httpcore2/_async/interfaces.py index f394d843..008d295d 100644 --- a/src/httpcore2/httpcore2/_async/interfaces.py +++ b/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. diff --git a/src/httpcore2/httpcore2/_sync/connection.py b/src/httpcore2/httpcore2/_sync/connection.py index 6c213cb4..5634ce99 100644 --- a/src/httpcore2/httpcore2/_sync/connection.py +++ b/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 diff --git a/src/httpcore2/httpcore2/_sync/connection_pool.py b/src/httpcore2/httpcore2/_sync/connection_pool.py index 287d9fcf..a6051bac 100644 --- a/src/httpcore2/httpcore2/_sync/connection_pool.py +++ b/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: diff --git a/src/httpcore2/httpcore2/_sync/http2.py b/src/httpcore2/httpcore2/_sync/http2.py index b0c42c71..992b42ce 100644 --- a/src/httpcore2/httpcore2/_sync/http2.py +++ b/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 diff --git a/src/httpcore2/httpcore2/_sync/interfaces.py b/src/httpcore2/httpcore2/_sync/interfaces.py index bbe7c7e6..0e66bd1e 100644 --- a/src/httpcore2/httpcore2/_sync/interfaces.py +++ b/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. diff --git a/tests/httpcore2/_async/test_connection_pool.py b/tests/httpcore2/_async/test_connection_pool.py index 832c9c1f..11b9a5d4 100644 --- a/tests/httpcore2/_async/test_connection_pool.py +++ b/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 diff --git a/tests/httpcore2/_sync/test_connection_pool.py b/tests/httpcore2/_sync/test_connection_pool.py index 137d4fd8..693b7371 100644 --- a/tests/httpcore2/_sync/test_connection_pool.py +++ b/tests/httpcore2/_sync/test_connection_pool.py @@ -789,3 +789,98 @@ def trace(name: str, kwargs: dict[str, typing.Any]) -> None: "http11.response_closed.started", "http11.response_closed.complete", ] + + + +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.ConnectionPool): + assign_passes = 0 + + def _assign_requests_to_connections(self) -> list[httpcore2.ConnectionInterface]: + CountingPool.assign_passes += 1 + return super()._assign_requests_to_connections() + + network_backend = httpcore2.MockBackend( + [ + 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 + ) + + def fetch(pool: httpcore2.ConnectionPool) -> None: + with pool.stream("GET", "https://example.com/") as response: + response.read() + assert response.status == 200 + + with CountingPool(max_connections=1, network_backend=network_backend) as pool: + 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 + + + +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.ConnectionPool): + max_queued_after_pass = 0 + + def _assign_requests_to_connections(self) -> list[httpcore2.ConnectionInterface]: + 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.MockBackend( + buffer=[ + hyperframe.frame.SettingsFrame().serialize(), + *response_frames(1), + *response_frames(3), + *response_frames(5), + *response_frames(7), + ], + http2=True, + ) + + def fetch(pool: httpcore2.ConnectionPool) -> None: + response = pool.request("GET", "https://example.com/") + assert response.status == 200 + + with QueueObservingPool(network_backend=network_backend, max_connections=1, http2=True) as pool: + # Warm the connection; it returns to the pool IDLE. + fetch(pool) + with concurrency.open_nursery() as nursery: + for _ in range(3): + nursery.start_soon(fetch, pool) + + assert QueueObservingPool.max_queued_after_pass == 0