diff --git a/CHANGELOG.md b/CHANGELOG.md index 5761a3cd..aae1c9bf 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/). ## Unreleased +- Improve connection pool performance when many requests are queued, by scanning the pool's connections once per assignment pass instead of once per queued request. - Fix `max_keepalive_connections` not being properly handled. (#1000) ## Version 1.0.9 (April 24th, 2025) diff --git a/httpcore/_async/connection_pool.py b/httpcore/_async/connection_pool.py index 5ef74e64..4b0d5abc 100644 --- a/httpcore/_async/connection_pool.py +++ b/httpcore/_async/connection_pool.py @@ -300,15 +300,22 @@ def _assign_requests_to_connections(self) -> list[AsyncConnectionInterface]: # Assign queued requests to connections. queued_requests = [request for request in self._requests if request.is_queued()] + if not queued_requests: + return closing_connections + + # Which connections are available or idle does not depend on the request, + # so scan the pool once and keep both lists up to date as the loop below + # creates and closes connections, rather than re-scanning every connection + # for every queued request. + available = [c for c in self._connections if c.is_available()] + idle_connections = [c for c in self._connections if c.is_idle()] + for pool_request in queued_requests: origin = pool_request.request.url.origin available_connections = [ connection - for connection in self._connections - if connection.can_handle_request(origin) and connection.is_available() - ] - idle_connections = [ - connection for connection in self._connections if connection.is_idle() + for connection in available + if connection.can_handle_request(origin) ] # There are three cases for how we may be able to handle the request: @@ -325,16 +332,27 @@ def _assign_requests_to_connections(self) -> list[AsyncConnectionInterface]: # log: "creating new connection" connection = self.create_connection(origin) self._connections.append(connection) + if connection.is_available(): + available.append(connection) pool_request.assign_to_connection(connection) elif idle_connections: # log: "closing idle connection" - connection = idle_connections[0] + connection = idle_connections.pop(0) self._connections.remove(connection) + if connection in available: + available.remove(connection) closing_connections.append(connection) # log: "creating new connection" connection = self.create_connection(origin) self._connections.append(connection) + if connection.is_available(): + available.append(connection) pool_request.assign_to_connection(connection) + elif not available: + # The pool is full, nothing is idle, and no connection is + # available for any origin, so no later request in the queue + # can be assigned either. + break return closing_connections diff --git a/httpcore/_sync/connection_pool.py b/httpcore/_sync/connection_pool.py index 4b26f9c6..096fc55d 100644 --- a/httpcore/_sync/connection_pool.py +++ b/httpcore/_sync/connection_pool.py @@ -300,15 +300,22 @@ def _assign_requests_to_connections(self) -> list[ConnectionInterface]: # Assign queued requests to connections. queued_requests = [request for request in self._requests if request.is_queued()] + if not queued_requests: + return closing_connections + + # Which connections are available or idle does not depend on the request, + # so scan the pool once and keep both lists up to date as the loop below + # creates and closes connections, rather than re-scanning every connection + # for every queued request. + available = [c for c in self._connections if c.is_available()] + idle_connections = [c for c in self._connections if c.is_idle()] + for pool_request in queued_requests: origin = pool_request.request.url.origin available_connections = [ connection - for connection in self._connections - if connection.can_handle_request(origin) and connection.is_available() - ] - idle_connections = [ - connection for connection in self._connections if connection.is_idle() + for connection in available + if connection.can_handle_request(origin) ] # There are three cases for how we may be able to handle the request: @@ -325,16 +332,27 @@ def _assign_requests_to_connections(self) -> list[ConnectionInterface]: # log: "creating new connection" connection = self.create_connection(origin) self._connections.append(connection) + if connection.is_available(): + available.append(connection) pool_request.assign_to_connection(connection) elif idle_connections: # log: "closing idle connection" - connection = idle_connections[0] + connection = idle_connections.pop(0) self._connections.remove(connection) + if connection in available: + available.remove(connection) closing_connections.append(connection) # log: "creating new connection" connection = self.create_connection(origin) self._connections.append(connection) + if connection.is_available(): + available.append(connection) pool_request.assign_to_connection(connection) + elif not available: + # The pool is full, nothing is idle, and no connection is + # available for any origin, so no later request in the queue + # can be assigned either. + break return closing_connections diff --git a/tests/_async/test_connection_pool.py b/tests/_async/test_connection_pool.py index bc4b251e..cd3ca201 100644 --- a/tests/_async/test_connection_pool.py +++ b/tests/_async/test_connection_pool.py @@ -835,3 +835,39 @@ async def trace(name, kwargs): "http11.response_closed.started", "http11.response_closed.complete", ] + + +@pytest.mark.anyio +async def test_connection_pool_closes_idle_connection_for_new_origin(): + """ + A pool at 'max_connections' with an idle connection to one origin should close + that connection to make room for a request to a different origin. + """ + network_backend = httpcore.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!", + ] + ) + + async with httpcore.AsyncConnectionPool( + network_backend=network_backend, max_connections=1, http2=True + ) as pool: + response = await pool.request("GET", "https://example.com/") + assert response.status == 200 + info = [repr(c) for c in pool.connections] + assert info == [ + "" + ] + + # A request to a different origin can not reuse the idle connection, and + # the pool is full, so the idle connection is closed and replaced. + response = await pool.request("GET", "https://other.com/") + assert response.status == 200 + info = [repr(c) for c in pool.connections] + assert info == [ + "" + ] diff --git a/tests/_sync/test_connection_pool.py b/tests/_sync/test_connection_pool.py index 7adc3f5c..029587d9 100644 --- a/tests/_sync/test_connection_pool.py +++ b/tests/_sync/test_connection_pool.py @@ -835,3 +835,39 @@ def trace(name, kwargs): "http11.response_closed.started", "http11.response_closed.complete", ] + + + +def test_connection_pool_closes_idle_connection_for_new_origin(): + """ + A pool at 'max_connections' with an idle connection to one origin should close + that connection to make room for a request to a different origin. + """ + network_backend = httpcore.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!", + ] + ) + + with httpcore.ConnectionPool( + network_backend=network_backend, max_connections=1, http2=True + ) as pool: + response = pool.request("GET", "https://example.com/") + assert response.status == 200 + info = [repr(c) for c in pool.connections] + assert info == [ + "" + ] + + # A request to a different origin can not reuse the idle connection, and + # the pool is full, so the idle connection is closed and replaced. + response = pool.request("GET", "https://other.com/") + assert response.status == 200 + info = [repr(c) for c in pool.connections] + assert info == [ + "" + ]