Fix ConnectionPool deadlock from reentrant _optional_thread_lock

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)

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)

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()

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")

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()