From a3e1a2bbc27c67e752a1142484360d6a058a70df Mon Sep 17 00:00:00 2001 From: Marcelo Trylesinski Date: Thu, 23 Jul 2026 09:18:30 +0200 Subject: [PATCH 1/3] Assign each released connection to a single queued request Assignment does not change an HTTP/1.1 connection's state, so a newly idle connection was handed to every queued request in one pass and re-picked by later passes until the winner sent on it. Every loser woke up, failed with ConnectionNotAvailable, re-entered the queue, and triggered another full O(n) assignment pass - quadratic churn at high queue depth. Drop a connection from the availability snapshot once assigned, and exclude idle connections already reserved by an in-flight request when building the snapshot. 1000 concurrent requests against a local server drop from 5.1s to 0.9s with the default pool, and from 100s to 1.1s with max_connections=1. HTTP/2 multiplexing is unaffected: an active h2 connection is not idle, so it stays available to additional streams. --- .../httpcore2/_async/connection_pool.py | 17 +++++++- .../httpcore2/_sync/connection_pool.py | 17 +++++++- .../httpcore2/_async/test_connection_pool.py | 43 +++++++++++++++++++ tests/httpcore2/_sync/test_connection_pool.py | 43 +++++++++++++++++++ 4 files changed, 116 insertions(+), 4 deletions(-) diff --git a/src/httpcore2/httpcore2/_async/connection_pool.py b/src/httpcore2/httpcore2/_async/connection_pool.py index 7011176e..20b72cb5 100644 --- a/src/httpcore2/httpcore2/_async/connection_pool.py +++ b/src/httpcore2/httpcore2/_async/connection_pool.py @@ -303,7 +303,16 @@ 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`. + available_connections = [ + connection + for connection in self._connections + if connection.is_available() and not (connection.is_idle() and connection in request_connections) + ] new_connection_budget = self._max_connections - len(self._connections) # Assign queued requests to connections. @@ -318,9 +327,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(): + # An HTTP/1.1 connection (or an idle HTTP/2 one) 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/connection_pool.py b/src/httpcore2/httpcore2/_sync/connection_pool.py index 287d9fcf..f09b7146 100644 --- a/src/httpcore2/httpcore2/_sync/connection_pool.py +++ b/src/httpcore2/httpcore2/_sync/connection_pool.py @@ -303,7 +303,16 @@ 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`. + available_connections = [ + connection + for connection in self._connections + if connection.is_available() and not (connection.is_idle() and connection in request_connections) + ] new_connection_budget = self._max_connections - len(self._connections) # Assign queued requests to connections. @@ -318,9 +327,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(): + # An HTTP/1.1 connection (or an idle HTTP/2 one) can + # only take this single request until it is released. + del available_connections[idx] break else: if new_connection_budget > 0: diff --git a/tests/httpcore2/_async/test_connection_pool.py b/tests/httpcore2/_async/test_connection_pool.py index 832c9c1f..2e38892a 100644 --- a/tests/httpcore2/_async/test_connection_pool.py +++ b/tests/httpcore2/_async/test_connection_pool.py @@ -789,3 +789,46 @@ 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 diff --git a/tests/httpcore2/_sync/test_connection_pool.py b/tests/httpcore2/_sync/test_connection_pool.py index 137d4fd8..022341b9 100644 --- a/tests/httpcore2/_sync/test_connection_pool.py +++ b/tests/httpcore2/_sync/test_connection_pool.py @@ -789,3 +789,46 @@ 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 From b9f33ba5f886a2e947f6c9d93efacdd7aa3def93 Mon Sep 17 00:00:00 2001 From: Marcelo Trylesinski Date: Thu, 23 Jul 2026 09:42:07 +0200 Subject: [PATCH 2/3] Exempt multiplexing connections from single-assignment reservation An idle HTTP/2 connection can serve further requests while reserved, so treating it like HTTP/1.1 could leave a queued burst waiting for the next pool event instead of multiplexing. Add can_multiplex() to the connection interface (False by default, True for established HTTP/2) and only apply the reserved-idle exclusion to connections that cannot multiplex. --- src/httpcore2/httpcore2/_async/connection.py | 3 ++ .../httpcore2/_async/connection_pool.py | 12 +++-- src/httpcore2/httpcore2/_async/http2.py | 3 ++ src/httpcore2/httpcore2/_async/interfaces.py | 10 ++++ src/httpcore2/httpcore2/_sync/connection.py | 3 ++ .../httpcore2/_sync/connection_pool.py | 12 +++-- src/httpcore2/httpcore2/_sync/http2.py | 3 ++ src/httpcore2/httpcore2/_sync/interfaces.py | 10 ++++ .../httpcore2/_async/test_connection_pool.py | 52 +++++++++++++++++++ tests/httpcore2/_sync/test_connection_pool.py | 52 +++++++++++++++++++ 10 files changed, 150 insertions(+), 10 deletions(-) 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 20b72cb5..e254fd41 100644 --- a/src/httpcore2/httpcore2/_async/connection_pool.py +++ b/src/httpcore2/httpcore2/_async/connection_pool.py @@ -307,11 +307,13 @@ def _assign_requests_to_connections(self) -> list[AsyncConnectionInterface]: # 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`. + # 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) + 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) @@ -330,9 +332,9 @@ def _assign_requests_to_connections(self) -> list[AsyncConnectionInterface]: for idx, connection in enumerate(available_connections): if connection.can_handle_request(origin): pool_request.assign_to_connection(connection) - if connection.is_idle(): - # An HTTP/1.1 connection (or an idle HTTP/2 one) can - # only take this single request until it is released. + 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: 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 f09b7146..20912e5c 100644 --- a/src/httpcore2/httpcore2/_sync/connection_pool.py +++ b/src/httpcore2/httpcore2/_sync/connection_pool.py @@ -307,11 +307,13 @@ def _assign_requests_to_connections(self) -> list[ConnectionInterface]: # 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`. + # 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) + 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) @@ -330,9 +332,9 @@ def _assign_requests_to_connections(self) -> list[ConnectionInterface]: for idx, connection in enumerate(available_connections): if connection.can_handle_request(origin): pool_request.assign_to_connection(connection) - if connection.is_idle(): - # An HTTP/1.1 connection (or an idle HTTP/2 one) can - # only take this single request until it is released. + 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: 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 2e38892a..11b9a5d4 100644 --- a/tests/httpcore2/_async/test_connection_pool.py +++ b/tests/httpcore2/_async/test_connection_pool.py @@ -832,3 +832,55 @@ async def fetch(pool: httpcore2.AsyncConnectionPool) -> None: # 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 022341b9..693b7371 100644 --- a/tests/httpcore2/_sync/test_connection_pool.py +++ b/tests/httpcore2/_sync/test_connection_pool.py @@ -832,3 +832,55 @@ def fetch(pool: httpcore2.ConnectionPool) -> None: # 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 From 0ed00a0b5130a6ca1a24aac2f8ce7d89d9c27546 Mon Sep 17 00:00:00 2001 From: Marcelo Trylesinski Date: Thu, 23 Jul 2026 10:43:42 +0200 Subject: [PATCH 3/3] Stop scanning the request queue once no connection can be assigned Each assignment pass walked every in-flight request even when the pool was saturated, and re-probed reserved idle connections for expiry with an is_readable socket check on every interleaved pass. Break out of the assignment loop once no connection is available and no new one may be created, and skip expiry checks and surplus-keepalive eviction for connections reserved by an assigned request - they were health-checked at assignment time, and evicting them would hand the winning request a closed connection. 1000 unbounded concurrent requests against a local server now complete in 0.41s versus 0.51s sequential, compared to 0.93s before this change and 5.1s before #1075. --- .../httpcore2/_async/connection_pool.py | 29 +++++++++++++------ .../httpcore2/_sync/connection_pool.py | 29 +++++++++++++------ 2 files changed, 40 insertions(+), 18 deletions(-) diff --git a/src/httpcore2/httpcore2/_async/connection_pool.py b/src/httpcore2/httpcore2/_async/connection_pool.py index e254fd41..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: @@ -317,8 +323,13 @@ def _assign_requests_to_connections(self) -> list[AsyncConnectionInterface]: ] 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 diff --git a/src/httpcore2/httpcore2/_sync/connection_pool.py b/src/httpcore2/httpcore2/_sync/connection_pool.py index 20912e5c..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: @@ -317,8 +323,13 @@ def _assign_requests_to_connections(self) -> list[ConnectionInterface]: ] 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