Skip to content

Maintain connection reservations incrementally in the pool #1076

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Kludex wants to merge 7 commits into
base: main
Choose a base branch
from
Open

Maintain connection reservations incrementally in the pool #1076

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
a3e1a2b
Assign each released connection to a single queued request
Kludex Jul 23, 2026
b9f33ba
Exempt multiplexing connections from single-assignment reservation
Kludex Jul 23, 2026
0ed00a0
Stop scanning the request queue once no connection can be assigned
Kludex Jul 23, 2026
d6582d8
Maintain request-connection reservations incrementally
Kludex Jul 23, 2026
58b1b65
Fix sync connection reservation races
Kludex Aug 2, 2026
13f976f
Preserve HTTP/2 multiplexing through proxies
Kludex Aug 2, 2026
d718f50
Merge remote-tracking branch 'origin/main' into pool-incremental-rese…
Kludex Aug 3, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
50 changes: 36 additions & 14 deletions src/httpcore2/httpcore2/_async/connection_pool.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -112,6 +112,11 @@ def __init__(
self._connections: list[AsyncConnectionInterface] = []
self._requests: list[AsyncPoolRequest] = []

# Reference counts of connections held by in-flight requests,
# maintained incrementally so assignment passes never rebuild them
# by scanning the full request list.
self._request_connections: dict[AsyncConnectionInterface, int] = {}

# We only mutate the state of the connection pool within an 'optional_thread_lock'
# context. This holds a threading lock unless we're running in async mode,
# in which case it is a no-op.
Expand Down Expand Up @@ -227,14 +232,17 @@ async def handle_async_request(self, request: Request) -> Response:
# handle a request, but then become unavailable.
#
# In this case we clear the connection and try again.
pool_request.clear_connection()
with self._optional_thread_lock:
self._release_request_connection(pool_request)
pool_request.clear_connection()
else:
break # pragma: no cover

except BaseException as exc:
with self._optional_thread_lock:
# For any exception or cancellation we remove the request from
# the queue, and then re-assign requests to connections.
self._release_request_connection(pool_request)
self._requests.remove(pool_request)
closing = self._assign_requests_to_connections()

Expand All @@ -251,6 +259,19 @@ async def handle_async_request(self, request: Request) -> Response:
extensions=response.extensions,
)

def _reserve_connection(self, pool_request: AsyncPoolRequest, connection: AsyncConnectionInterface) -> None:
pool_request.assign_to_connection(connection)
self._request_connections[connection] = self._request_connections.get(connection, 0) + 1

def _release_request_connection(self, pool_request: AsyncPoolRequest) -> None:
connection = pool_request.connection
if connection is not None:
count = self._request_connections[connection] - 1
if count:
self._request_connections[connection] = count
else:
del self._request_connections[connection]

def _assign_requests_to_connections(self) -> list[AsyncConnectionInterface]:
"""
Manage the state of the connection pool, assigning incoming
Expand All @@ -267,7 +288,7 @@ def _assign_requests_to_connections(self) -> list[AsyncConnectionInterface]:
# 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}
request_connections = self._request_connections

# First we handle cleaning up any connections that are closed
# or have expired their keep-alive, in a single pass. Reserved
Expand Down Expand Up @@ -310,16 +331,16 @@ def _assign_requests_to_connections(self) -> list[AsyncConnectionInterface]:
# it per queued request — this is what brings the loop from O(N*M) to
# O(N+M) in the common case.
#
# 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.
# An established non-multiplexing connection already assigned to an
# in-flight request is reserved. Its state may transition from IDLE to
# ACTIVE after `is_available()` returns, so use `is_connected()` here
# rather than checking its mutable idle state. Multiplexing connections
# and not-yet-connected HTTP/2 candidates remain available.
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())
and not (connection.is_connected() and connection in request_connections and not connection.can_multiplex())

@macroscopeapp macroscopeapp Bot Aug 2, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟠 High _async/connection_pool.py:343

The new availability filter on line 343 excludes established connections from available_connections when can_multiplex() returns False. Proxy connection wrappers such as AsyncTunnelHTTPConnection and AsyncSocks5Connection do not override the newly added can_multiplex() method and inherit the default False from AsyncConnectionInterface, even when their inner connection has negotiated HTTP/2. As a result, once the first proxied HTTP/2 request reserves such a connection, concurrent requests cannot reuse the established HTTP/2 connection and instead queue or open extra proxy connections, defeating multiplexing. The proxy wrappers need to delegate can_multiplex() to their underlying connection so the pool filter recognizes them as multiplexing-capable.

Also found in 2 other location(s)

src/httpcore2/httpcore2/_sync/connection_pool.py:343

The new availability filter treats every proxy wrapper that inherits the default ConnectionInterface.can_multiplex() as non-multiplexing. TunnelHTTPConnection and Socks5Connection can replace their inner connection with an HTTP2Connection, but neither delegates can_multiplex() to it. Once such a proxy connection is established and reserved by one request, line 343 excludes it from available_connections, so concurrent requests cannot reuse the HTTP/2 connection and may queue or open extra proxy connections instead of multiplexing.

src/httpcore2/httpcore2/_sync/connection_pool.py:343

available_connections excludes every reserved established connection whose outer class does not override can_multiplex(). Proxy connection wrappers such as TunnelHTTPConnection and Socks5Connection can contain an HTTP2Connection but inherit the new default False, so after the first proxied HTTP/2 request is reserved, concurrent requests cannot reuse that HTTP/2 connection and instead queue or open extra connections. The wrapper needs to delegate multiplex capability to its established inner connection.

🚀 Reply "fix it for me" or copy this AI Prompt for your agent: 
In file @src/httpcore2/httpcore2/_async/connection_pool.py around line 343:

The new availability filter on line 343 excludes established connections from `available_connections` when `can_multiplex()` returns `False`. Proxy connection wrappers such as `AsyncTunnelHTTPConnection` and `AsyncSocks5Connection` do not override the newly added `can_multiplex()` method and inherit the default `False` from `AsyncConnectionInterface`, even when their inner connection has negotiated HTTP/2. As a result, once the first proxied HTTP/2 request reserves such a connection, concurrent requests cannot reuse the established HTTP/2 connection and instead queue or open extra proxy connections, defeating multiplexing. The proxy wrappers need to delegate `can_multiplex()` to their underlying connection so the pool filter recognizes them as multiplexing-capable.

Also found in 2 other location(s):
- src/httpcore2/httpcore2/_sync/connection_pool.py:343 -- The new availability filter treats every proxy wrapper that inherits the default `ConnectionInterface.can_multiplex()` as non-multiplexing. `TunnelHTTPConnection` and `Socks5Connection` can replace their inner connection with an `HTTP2Connection`, but neither delegates `can_multiplex()` to it. Once such a proxy connection is established and reserved by one request, line 343 excludes it from `available_connections`, so concurrent requests cannot reuse the HTTP/2 connection and may queue or open extra proxy connections instead of multiplexing.
- src/httpcore2/httpcore2/_sync/connection_pool.py:343 -- `available_connections` excludes every reserved established connection whose outer class does not override `can_multiplex()`. Proxy connection wrappers such as `TunnelHTTPConnection` and `Socks5Connection` can contain an `HTTP2Connection` but inherit the new default `False`, so after the first proxied HTTP/2 request is reserved, concurrent requests cannot reuse that HTTP/2 connection and instead queue or open extra connections. The wrapper needs to delegate multiplex capability to its established inner connection.

@cubic-dev-ai cubic-dev-ai Bot Aug 2, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1: Concurrent HTTP/2 requests through an HTTP CONNECT or SOCKS proxy no longer share an established tunnel: these wrappers report can_multiplex() == False even after negotiating AsyncHTTP2Connection, so this predicate excludes the active connection. Forward can_multiplex() from both proxy wrappers (or otherwise base this reservation check on the negotiated delegate) before applying the non-multiplexing reservation rule.

Prompt for AI agents 
Check if this issue is valid — if so, understand the root cause and fix it. At src/httpcore2/httpcore2/_async/connection_pool.py, line 343:

<comment>Concurrent HTTP/2 requests through an HTTP CONNECT or SOCKS proxy no longer share an established tunnel: these wrappers report `can_multiplex() == False` even after negotiating `AsyncHTTP2Connection`, so this predicate excludes the active connection. Forward `can_multiplex()` from both proxy wrappers (or otherwise base this reservation check on the negotiated delegate) before applying the non-multiplexing reservation rule.</comment>

<file context>
@@ -331,16 +331,16 @@ def _assign_requests_to_connections(self) -> list[AsyncConnectionInterface]:
             for connection in self._connections
             if connection.is_available()
-            and not (connection.is_idle() and connection in request_connections and not connection.can_multiplex())
+            and not (connection.is_connected() and connection in request_connections and not connection.can_multiplex())
         ]
         new_connection_budget = self._max_connections - len(self._connections)
</file context>

]
new_connection_budget = self._max_connections - len(self._connections)

Expand All @@ -342,17 +363,17 @@ def _assign_requests_to_connections(self) -> list[AsyncConnectionInterface]:
# to handle the request.
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.
if connection.is_connected() and not connection.can_multiplex():
# Remove an established HTTP/1.1 connection before
# waking the request, which may transition it to ACTIVE.
del available_connections[idx]
self._reserve_connection(pool_request, connection)
break
else:
if new_connection_budget > 0:
connection = self.create_connection(origin)
self._connections.append(connection)
pool_request.assign_to_connection(connection)
self._reserve_connection(pool_request, connection)
new_connection_budget -= 1
continue
for idx, connection in enumerate(available_connections):
Expand All @@ -362,7 +383,7 @@ def _assign_requests_to_connections(self) -> list[AsyncConnectionInterface]:
closing_connections.append(connection)
connection = self.create_connection(origin)
self._connections.append(connection)
pool_request.assign_to_connection(connection)
self._reserve_connection(pool_request, connection)
break

return closing_connections
Expand Down Expand Up @@ -434,6 +455,7 @@ async def aclose(self) -> None:
await self._stream.aclose()

with self._pool._optional_thread_lock:
self._pool._release_request_connection(self._pool_request)
self._pool._requests.remove(self._pool_request)
closing = self._pool._assign_requests_to_connections()

Expand Down
6 changes: 6 additions & 0 deletions src/httpcore2/httpcore2/_async/http_proxy.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -216,6 +216,9 @@ def has_expired(self) -> bool:
def is_idle(self) -> bool:
return self._connection.is_idle()

def can_multiplex(self) -> bool:
return self._connection.can_multiplex()

def is_closed(self) -> bool:
return self._connection.is_closed()

Expand Down Expand Up @@ -345,6 +348,9 @@ def has_expired(self) -> bool:
def is_idle(self) -> bool:
return self._connection.is_idle()

def can_multiplex(self) -> bool:
return self._connection.can_multiplex()

def is_closed(self) -> bool:
return self._connection.is_closed()

Expand Down
3 changes: 3 additions & 0 deletions src/httpcore2/httpcore2/_async/socks_proxy.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -311,6 +311,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: # pragma: no cover
return self._connect_failed
Expand Down
50 changes: 36 additions & 14 deletions src/httpcore2/httpcore2/_sync/connection_pool.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -112,6 +112,11 @@ def __init__(
self._connections: list[ConnectionInterface] = []
self._requests: list[PoolRequest] = []

# Reference counts of connections held by in-flight requests,
# maintained incrementally so assignment passes never rebuild them
# by scanning the full request list.
self._request_connections: dict[ConnectionInterface, int] = {}

# We only mutate the state of the connection pool within an 'optional_thread_lock'
# context. This holds a threading lock unless we're running in async mode,
# in which case it is a no-op.
Expand Down Expand Up @@ -227,14 +232,17 @@ def handle_request(self, request: Request) -> Response:
# handle a request, but then become unavailable.
#
# In this case we clear the connection and try again.
pool_request.clear_connection()
with self._optional_thread_lock:
self._release_request_connection(pool_request)
pool_request.clear_connection()
else:
break # pragma: no cover

except BaseException as exc:
with self._optional_thread_lock:
# For any exception or cancellation we remove the request from
# the queue, and then re-assign requests to connections.
self._release_request_connection(pool_request)
self._requests.remove(pool_request)
closing = self._assign_requests_to_connections()

Expand All @@ -251,6 +259,19 @@ def handle_request(self, request: Request) -> Response:
extensions=response.extensions,
)

def _reserve_connection(self, pool_request: PoolRequest, connection: ConnectionInterface) -> None:
pool_request.assign_to_connection(connection)
self._request_connections[connection] = self._request_connections.get(connection, 0) + 1

def _release_request_connection(self, pool_request: PoolRequest) -> None:
connection = pool_request.connection
if connection is not None:
count = self._request_connections[connection] - 1
if count:
self._request_connections[connection] = count
else:
del self._request_connections[connection]

def _assign_requests_to_connections(self) -> list[ConnectionInterface]:
"""
Manage the state of the connection pool, assigning incoming
Expand All @@ -267,7 +288,7 @@ def _assign_requests_to_connections(self) -> list[ConnectionInterface]:
# 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}
request_connections = self._request_connections

# First we handle cleaning up any connections that are closed
# or have expired their keep-alive, in a single pass. Reserved
Expand Down Expand Up @@ -310,16 +331,16 @@ def _assign_requests_to_connections(self) -> list[ConnectionInterface]:
# it per queued request — this is what brings the loop from O(N*M) to
# O(N+M) in the common case.
#
# 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.
# An established non-multiplexing connection already assigned to an
# in-flight request is reserved. Its state may transition from IDLE to
# ACTIVE after `is_available()` returns, so use `is_connected()` here
# rather than checking its mutable idle state. Multiplexing connections
# and not-yet-connected HTTP/2 candidates remain available.
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())
and not (connection.is_connected() and connection in request_connections and not connection.can_multiplex())
]
new_connection_budget = self._max_connections - len(self._connections)

Expand All @@ -342,17 +363,17 @@ def _assign_requests_to_connections(self) -> list[ConnectionInterface]:
# to handle the request.
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.
if connection.is_connected() and not connection.can_multiplex():
# Remove an established HTTP/1.1 connection before
# waking the request, which may transition it to ACTIVE.
del available_connections[idx]
self._reserve_connection(pool_request, connection)
break
else:
if new_connection_budget > 0:
connection = self.create_connection(origin)
self._connections.append(connection)
pool_request.assign_to_connection(connection)
self._reserve_connection(pool_request, connection)
new_connection_budget -= 1
continue
for idx, connection in enumerate(available_connections):
Expand All @@ -362,7 +383,7 @@ def _assign_requests_to_connections(self) -> list[ConnectionInterface]:
closing_connections.append(connection)
connection = self.create_connection(origin)
self._connections.append(connection)
pool_request.assign_to_connection(connection)
self._reserve_connection(pool_request, connection)
break

return closing_connections
Expand Down Expand Up @@ -434,6 +455,7 @@ def close(self) -> None:
self._stream.close()

with self._pool._optional_thread_lock:
self._pool._release_request_connection(self._pool_request)
self._pool._requests.remove(self._pool_request)
closing = self._pool._assign_requests_to_connections()

Expand Down
6 changes: 6 additions & 0 deletions src/httpcore2/httpcore2/_sync/http_proxy.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -216,6 +216,9 @@ def has_expired(self) -> bool:
def is_idle(self) -> bool:
return self._connection.is_idle()

def can_multiplex(self) -> bool:
return self._connection.can_multiplex()

def is_closed(self) -> bool:
return self._connection.is_closed()

Expand Down Expand Up @@ -345,6 +348,9 @@ def has_expired(self) -> bool:
def is_idle(self) -> bool:
return self._connection.is_idle()

def can_multiplex(self) -> bool:
return self._connection.can_multiplex()

def is_closed(self) -> bool:
return self._connection.is_closed()

Expand Down
3 changes: 3 additions & 0 deletions src/httpcore2/httpcore2/_sync/socks_proxy.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -311,6 +311,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: # pragma: no cover
return self._connect_failed
Expand Down
Loading
Loading