From e098b6f56ed7832925b6afae82ceef6e2755846b Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 8 Sep 2026 00:43:00 +0000 Subject: [PATCH] Fix ConnectionPool deadlock from reentrant _optional_thread_lock PoolByteStream.__iter__ is a generator. If it is abandoned mid-iteration (the last reference to a response/stream is dropped without reading it to completion or calling .close()), CPython finalizes the generator by throwing GeneratorExit at its suspension point, which lands in PoolByteStream.__iter__'s `except BaseException` handler and calls self.close() -> re-acquires ConnectionPool._optional_thread_lock. In production this finalization is triggered by CPython's garbage collector, which can run at essentially any allocation point, including from inside _assign_requests_to_connections() -- which only ever runs while _optional_thread_lock is already held. If that happens on the same thread, the plain (non-reentrant) lock deadlocks permanently. Reported independently by multiple users of multithreaded servers (frequently via the OpenAI SDK, which sits on httpx/httpcore) in #990. This makes ThreadLock's underlying lock a threading.RLock so the same thread can safely re-enter it, and hardens the three list.remove(connection) call sites in _assign_requests_to_connections() that would otherwise raise ValueError if a reentrant call had already removed the same connection -- a state-corruption risk that a bare Lock -> RLock swap would introduce silently. This is sync-only: AsyncThreadLock is already a no-op, so no equivalent scenario exists in async mode. Adds a deterministic regression test (no reliance on GC timing) that reproduces the original deadlock by simulating the exact reentrant interleaving, plus direct unit tests of ThreadLock's reentrancy contract. Fixes #990 Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01RnkMYMmQtTsupovqXuxDaw --- httpcore/_async/connection_pool.py | 26 +++++-- httpcore/_sync/connection_pool.py | 26 +++++-- httpcore/_synchronization.py | 13 +++- tests/_sync/test_connection_pool.py | 113 ++++++++++++++++++++++++++++ tests/test_synchronization.py | 58 ++++++++++++++ 5 files changed, 221 insertions(+), 15 deletions(-) create mode 100644 tests/test_synchronization.py diff --git a/httpcore/_async/connection_pool.py b/httpcore/_async/connection_pool.py index 5ef74e64..488a8771 100644 --- a/httpcore/_async/connection_pool.py +++ b/httpcore/_async/connection_pool.py @@ -281,22 +281,33 @@ def _assign_requests_to_connections(self) -> list[AsyncConnectionInterface]: # First we handle cleaning up any connections that are closed, # have expired their keep-alive, or surplus idle connections. + # + # Note: calling into `is_closed()` / `has_expired()` / `is_idle()` below + # can, via garbage collection of an abandoned `PoolByteStream`, trigger a + # *reentrant* call into this same method on this same thread (see the + # docstring on `ThreadLock`). That reentrant call may already have + # removed a connection from `self._connections` by the time we get back + # here, so every removal below is guarded rather than assumed to + # succeed. for connection in list(self._connections): if connection.is_closed(): # log: "removing closed connection" - self._connections.remove(connection) + if connection in self._connections: + self._connections.remove(connection) elif connection.has_expired(): # log: "closing expired connection" - self._connections.remove(connection) - closing_connections.append(connection) + if connection in self._connections: + self._connections.remove(connection) + closing_connections.append(connection) elif ( connection.is_idle() and sum(connection.is_idle() for connection in self._connections) > self._max_keepalive_connections ): # log: "closing idle connection" - self._connections.remove(connection) - closing_connections.append(connection) + if connection in self._connections: + self._connections.remove(connection) + closing_connections.append(connection) # Assign queued requests to connections. queued_requests = [request for request in self._requests if request.is_queued()] @@ -329,8 +340,9 @@ def _assign_requests_to_connections(self) -> list[AsyncConnectionInterface]: elif idle_connections: # log: "closing idle connection" connection = idle_connections[0] - self._connections.remove(connection) - closing_connections.append(connection) + if connection in self._connections: + self._connections.remove(connection) + closing_connections.append(connection) # log: "creating new connection" connection = self.create_connection(origin) self._connections.append(connection) diff --git a/httpcore/_sync/connection_pool.py b/httpcore/_sync/connection_pool.py index 4b26f9c6..8f8629af 100644 --- a/httpcore/_sync/connection_pool.py +++ b/httpcore/_sync/connection_pool.py @@ -281,22 +281,33 @@ def _assign_requests_to_connections(self) -> list[ConnectionInterface]: # First we handle cleaning up any connections that are closed, # have expired their keep-alive, or surplus idle connections. + # + # Note: calling into `is_closed()` / `has_expired()` / `is_idle()` below + # can, via garbage collection of an abandoned `PoolByteStream`, trigger a + # *reentrant* call into this same method on this same thread (see the + # docstring on `ThreadLock`). That reentrant call may already have + # removed a connection from `self._connections` by the time we get back + # here, so every removal below is guarded rather than assumed to + # succeed. for connection in list(self._connections): if connection.is_closed(): # log: "removing closed connection" - self._connections.remove(connection) + if connection in self._connections: + self._connections.remove(connection) elif connection.has_expired(): # log: "closing expired connection" - self._connections.remove(connection) - closing_connections.append(connection) + if connection in self._connections: + self._connections.remove(connection) + closing_connections.append(connection) elif ( connection.is_idle() and sum(connection.is_idle() for connection in self._connections) > self._max_keepalive_connections ): # log: "closing idle connection" - self._connections.remove(connection) - closing_connections.append(connection) + if connection in self._connections: + self._connections.remove(connection) + closing_connections.append(connection) # Assign queued requests to connections. queued_requests = [request for request in self._requests if request.is_queued()] @@ -329,8 +340,9 @@ def _assign_requests_to_connections(self) -> list[ConnectionInterface]: elif idle_connections: # log: "closing idle connection" connection = idle_connections[0] - self._connections.remove(connection) - closing_connections.append(connection) + if connection in self._connections: + self._connections.remove(connection) + closing_connections.append(connection) # log: "creating new connection" connection = self.create_connection(origin) self._connections.append(connection) diff --git a/httpcore/_synchronization.py b/httpcore/_synchronization.py index 2ecc9e9c..440a6246 100644 --- a/httpcore/_synchronization.py +++ b/httpcore/_synchronization.py @@ -259,10 +259,21 @@ class ThreadLock: In the sync case `ThreadLock` provides thread locking. In the async case `AsyncThreadLock` is a no-op. + + This must be re-entrant (safe for the *same* thread to acquire it more + than once). `ConnectionPool` uses this lock around its internal + bookkeeping (`_assign_requests_to_connections`), which calls out to + connection/stream methods such as `is_idle()`. If the *last* reference to + an unclosed `PoolByteStream` is dropped while that bookkeeping is running + (for example because it's garbage collected), Python finalizes the + generator in `PoolByteStream.__iter__`, which calls `PoolByteStream.close()`, + which re-enters this same lock, on the same thread that is already + holding it. With a plain, non-reentrant lock this deadlocks the pool + permanently (see https://github.com/encode/httpcore/discussions/990). """ def __init__(self) -> None: - self._lock = threading.Lock() + self._lock = threading.RLock() def __enter__(self) -> ThreadLock: self._lock.acquire() diff --git a/tests/_sync/test_connection_pool.py b/tests/_sync/test_connection_pool.py index 7adc3f5c..d4b6ddc8 100644 --- a/tests/_sync/test_connection_pool.py +++ b/tests/_sync/test_connection_pool.py @@ -835,3 +835,116 @@ def trace(name, kwargs): "http11.response_closed.started", "http11.response_closed.complete", ] + + +def test_connection_pool_reentrant_close_does_not_deadlock(): + """ + Regression test for https://github.com/encode/httpcore/discussions/990 + + `PoolByteStream.__iter__` is a generator. If it is abandoned mid-iteration + (the last reference to the response/stream is dropped without reading it + to completion or calling `.close()`), CPython finalizes the generator by + throwing `GeneratorExit` at its suspension point. That lands in the + `except BaseException` handler in `PoolByteStream.__iter__`, which calls + `self.close()` -> re-acquires `ConnectionPool._optional_thread_lock`. + + In production reports (multithreaded servers using httpx/httpcore, + frequently via the OpenAI SDK) this finalization is triggered by + CPython's garbage collector, which can run at essentially any allocation + point -- including from *inside* `_assign_requests_to_connections()`, + which only ever runs while `_optional_thread_lock` is already held. If + that happens on the same thread, a non-reentrant lock deadlocks + permanently. + + This test reproduces the interleaving deterministically (without relying + on GC timing) by monkeypatching one connection's `is_idle()` -- a method + `_assign_requests_to_connections()` calls while holding the lock -- to + drop the last reference to an abandoned stream at exactly that point. + This is sync-only: in async mode `AsyncThreadLock` is a no-op, so there + is no lock to deadlock on, and no equivalent scenario exists. + """ + import threading + import types + + 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, ", + b"world!", + b"HTTP/1.1 200 OK\r\n", + b"Content-Type: plain/text\r\n", + b"Content-Length: 2\r\n", + b"\r\n", + b"OK", + ] + ) + + with httpcore.ConnectionPool( + network_backend=network_backend, max_connections=2 + ) as pool: + request1 = httpcore.Request( + "GET", "https://example.com/", headers=[(b"Host", b"example.com")] + ) + response1 = pool.handle_request(request1) + + # Partially consume the stream, then abandon it: the generator inside + # PoolByteStream.__iter__ is now suspended mid + # `for part in self._stream: yield part`, with its pool_request + # still active. + assert isinstance(response1.stream, typing.Iterable) + stream_iter = iter(response1.stream) + next(stream_iter) + + # Keep the only remaining reference to the generator in a mutable + # holder, so we can drop it from inside is_idle() below, simulating + # the moment CPython's GC would otherwise finalize it. + holder = [stream_iter] + del stream_iter, response1 + + assert len(pool.connections) == 1 + connection1 = pool.connections[0] + real_is_idle = connection1.is_idle + + def patched_is_idle(self: object) -> bool: + if holder: + # Drop the last reference to the abandoned generator *while + # this thread is inside _assign_requests_to_connections(), + # holding _optional_thread_lock*. This calls + # PoolByteStream.close(), which re-enters the same lock. + holder.clear() + return real_is_idle() + + connection1.is_idle = types.MethodType(patched_is_idle, connection1) # type: ignore[method-assign] + + # Sending a second request forces the pool to re-run + # _assign_requests_to_connections(), which calls connection1.is_idle() + # while holding the lock -> triggers the reentrant close() above. + # Run it on a background thread with a bounded join so that if this + # regresses, the test fails instead of hanging CI forever. + result: typing.Dict[str, object] = {} + + def worker() -> None: + try: + request2 = httpcore.Request( + "GET", + "https://example.com/", + headers=[(b"Host", b"example.com")], + ) + response2 = pool.handle_request(request2) + response2.read() + result["ok"] = True + except BaseException as exc: # pragma: no cover - diagnostic only + result["error"] = exc + + thread = threading.Thread(target=worker, daemon=True) + thread.start() + thread.join(timeout=5) + + assert not thread.is_alive(), ( + "ConnectionPool deadlocked: a reentrant call into " + "_optional_thread_lock did not resolve within 5 seconds." + ) + assert result.get("ok") is True, result.get("error") diff --git a/tests/test_synchronization.py b/tests/test_synchronization.py new file mode 100644 index 00000000..3bf1a95b --- /dev/null +++ b/tests/test_synchronization.py @@ -0,0 +1,58 @@ +import threading + +from httpcore._synchronization import ThreadLock + + +def test_thread_lock_is_reentrant(): + """ + `ThreadLock` guards `ConnectionPool`'s internal bookkeeping + (`_assign_requests_to_connections`), which calls out to connection and + stream methods. If the *last* reference to an unclosed `PoolByteStream` + is dropped while that bookkeeping is running (for example, because it is + garbage collected), `PoolByteStream.close()` re-enters this same lock on + the same thread that is already holding it. + + A plain, non-reentrant lock deadlocks permanently in that case. See + https://github.com/encode/httpcore/discussions/990 for real-world reports + of this happening in multithreaded servers. + + This test only checks the primitive's contract in isolation: that the + same thread can acquire the lock more than once without blocking. + """ + lock = ThreadLock() + + acquired_nested = threading.Event() + + with lock: + with lock: + acquired_nested.set() + + assert acquired_nested.is_set() + + +def test_thread_lock_still_blocks_other_threads(): + """ + Re-entrancy must not come at the cost of mutual exclusion between + *different* threads: a second thread must still block until the first + thread has released the lock completely. + """ + lock = ThreadLock() + other_thread_acquired = threading.Event() + release_first_thread = threading.Event() + + def other_thread_body() -> None: + with lock: + other_thread_acquired.set() + + with lock: + thread = threading.Thread(target=other_thread_body, daemon=True) + thread.start() + # The other thread should not be able to acquire the lock while + # we're still holding it. + assert not other_thread_acquired.wait(timeout=0.2) + release_first_thread.set() + + # Once we've released it, the other thread should be able to proceed. + thread.join(timeout=2) + assert not thread.is_alive() + assert other_thread_acquired.is_set()