From ca12dde9e88ef6a10818ad94886d8e6204401b89 Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Fri, 4 Sep 2026 06:32:55 -0500 Subject: [PATCH 1/7] PYTHON-6074 Fix pool deadlock when a greenlet is killed during checkin --- doc/changelog.rst | 4 ++ pymongo/asynchronous/pool.py | 114 ++++++++++++++++++++++--------- pymongo/synchronous/pool.py | 114 ++++++++++++++++++++++--------- test/asynchronous/test_client.py | 98 ++++++++++++++++++++++++++ test/test_client.py | 98 ++++++++++++++++++++++++++ 5 files changed, 360 insertions(+), 68 deletions(-) diff --git a/doc/changelog.rst b/doc/changelog.rst index f657b819a6..fa3bad6d5e 100644 --- a/doc/changelog.rst +++ b/doc/changelog.rst @@ -1043,6 +1043,9 @@ Bug fixes after non-resumable errors (`PYTHON-3389`_). - Fixed a bug where the client could be unable to discover the new primary after a simultaneous replica set election and reconfig (`PYTHON-2970`_). +- Fixed a bug where the synchronous client could permanently deadlock under + gevent when a greenlet was killed while checking a connection back into + the pool (`PYTHON-6074`_). Issues Resolved ............... @@ -1054,6 +1057,7 @@ in this release. .. _PYTHON-2484: https://jira.mongodb.org/browse/PYTHON-2484 .. _PYTHON-2970: https://jira.mongodb.org/browse/PYTHON-2970 .. _PYTHON-3389: https://jira.mongodb.org/browse/PYTHON-3389 +.. _PYTHON-6074: https://jira.mongodb.org/browse/PYTHON-6074 .. _PyMongo 4.3 release notes in JIRA: https://jira.mongodb.org/secure/ReleaseNote.jspa?projectId=10004&version=33425 Changes in Version 4.2.0 (2022/07/20) diff --git a/pymongo/asynchronous/pool.py b/pymongo/asynchronous/pool.py index b8dd042dc6..bb3336b2b3 100644 --- a/pymongo/asynchronous/pool.py +++ b/pymongo/asynchronous/pool.py @@ -1060,6 +1060,47 @@ async def _get_conn( return conn + def _checkin_apply( + self, conn: AsyncConnection, txn: bool, cursor: bool, forked: bool + ) -> tuple[Optional[str], bool, bool]: + """Apply checkin accounting and decide the connection's disposition. + + Caller must hold ``self.size_cond``. This method performs no cooperative + I/O (no await, no lock/condition acquire, no notify) so it is safe to + call while a gevent greenlet unwinds a ``GreenletExit``: gevent does not + re-throw ``GreenletExit`` at cooperative yields during unwind, so the + re-acquisition in the ``try/finally`` in ``checkin`` completes. The same + guarantee does not hold for an asyncio task cancelled with + ``CancelledError``, which is re-raised at the next await. Returns + ``(close_conn_reason, emit_closed, appended)`` describing work that must + be done outside the lock. See PYTHON-6074. + """ + self.active_contexts.discard(conn.cancel_context) + if txn: + self.ntxns -= 1 + elif cursor: + self.ncursors -= 1 + self.requests -= 1 + self.active_sockets -= 1 + self.operation_count -= 1 + close_conn_reason: Optional[str] = None + emit_closed = False + appended = False + if not forked: + if self.closed: + close_conn_reason = ConnectionClosedReason.POOL_CLOSED + elif conn.closed: + # CMAP requires the closed event be emitted after the check in. + emit_closed = True + elif self.stale_generation(conn.generation, conn.service_id): + close_conn_reason = ConnectionClosedReason.STALE + else: + conn.update_last_checkin_time() + conn.update_is_writable(bool(self.is_writable)) + self.conns.appendleft(conn) + appended = True + return close_conn_reason, emit_closed, appended + async def checkin(self, conn: AsyncConnection) -> None: """Return the connection to the pool, or if it's closed discard it. @@ -1071,44 +1112,49 @@ async def checkin(self, conn: AsyncConnection) -> None: conn.pinned_txn = False conn.pinned_cursor = False self._pinned_sockets.discard(conn) - async with self.lock: - self.active_contexts.discard(conn.cancel_context) + forked = self.pid != os.getpid() + # The pool accounting (requests/active_sockets) must be decremented and + # the connection returned under a single hold of self.size_cond. Under + # gevent, acquiring the lock and Condition.notify() both cooperatively + # yield, so a GreenletExit injected at such a yield -- e.g. a websocket + # handler greenlet being killed while it is checking a connection back + # in after a normal operation -- can interrupt checkin before the + # decrement, leaving requests/active_sockets permanently inflated and + # saturating the size gate (PYTHON-6074). The try/finally below + # re-applies the accounting while unwinding: gevent does not re-throw + # GreenletExit at cooperative yields during unwind, so the re-acquisition + # completes and the accounting is restored. + close_conn_reason: Optional[str] = None + emit_closed = False + accounted = False + try: + async with self.size_cond: + close_conn_reason, emit_closed, appended = self._checkin_apply( + conn, txn, cursor, forked + ) + accounted = True + if appended: + # Notify any threads waiting to create a connection. + self._max_connecting_cond.notify() + self.size_cond.notify() + finally: + if not accounted: + async with self.size_cond: + close_conn_reason, emit_closed, appended = self._checkin_apply( + conn, txn, cursor, forked + ) + if appended: + self._max_connecting_cond.notify() + self.size_cond.notify() telemetry = self._telemetry if telemetry._should_publish or (telemetry._log and _is_debug_enabled(_CONNECTION_LOGGER)): telemetry.checked_in(conn.id) - if self.pid != os.getpid(): + if emit_closed: + telemetry.connection_closed(conn.id, ConnectionClosedReason.ERROR) + if forked: await self.reset_without_pause() - else: - if self.closed: - await conn.close_conn(ConnectionClosedReason.POOL_CLOSED) - elif conn.closed: - # CMAP requires the closed event be emitted after the check in. - self._telemetry.connection_closed(conn.id, ConnectionClosedReason.ERROR) - else: - close_conn = False - async with self.lock: - # Hold the lock to ensure this section does not race with - # Pool.reset(). - if self.stale_generation(conn.generation, conn.service_id): - close_conn = True - else: - conn.update_last_checkin_time() - conn.update_is_writable(bool(self.is_writable)) - self.conns.appendleft(conn) - # Notify any threads waiting to create a connection. - self._max_connecting_cond.notify() - if close_conn: - await conn.close_conn(ConnectionClosedReason.STALE) - - async with self.size_cond: - if txn: - self.ntxns -= 1 - elif cursor: - self.ncursors -= 1 - self.requests -= 1 - self.active_sockets -= 1 - self.operation_count -= 1 - self.size_cond.notify() + elif close_conn_reason is not None: + await conn.close_conn(close_conn_reason) async def _perished(self, conn: AsyncConnection) -> bool: """Return True and close the connection if it is "perished". diff --git a/pymongo/synchronous/pool.py b/pymongo/synchronous/pool.py index e2a708e18f..28ff48d0bd 100644 --- a/pymongo/synchronous/pool.py +++ b/pymongo/synchronous/pool.py @@ -1056,6 +1056,47 @@ def _get_conn( return conn + def _checkin_apply( + self, conn: Connection, txn: bool, cursor: bool, forked: bool + ) -> tuple[Optional[str], bool, bool]: + """Apply checkin accounting and decide the connection's disposition. + + Caller must hold ``self.size_cond``. This method performs no cooperative + I/O (no await, no lock/condition acquire, no notify) so it is safe to + call while a gevent greenlet unwinds a ``GreenletExit``: gevent does not + re-throw ``GreenletExit`` at cooperative yields during unwind, so the + re-acquisition in the ``try/finally`` in ``checkin`` completes. The same + guarantee does not hold for an asyncio task cancelled with + ``CancelledError``, which is re-raised at the next await. Returns + ``(close_conn_reason, emit_closed, appended)`` describing work that must + be done outside the lock. See PYTHON-6074. + """ + self.active_contexts.discard(conn.cancel_context) + if txn: + self.ntxns -= 1 + elif cursor: + self.ncursors -= 1 + self.requests -= 1 + self.active_sockets -= 1 + self.operation_count -= 1 + close_conn_reason: Optional[str] = None + emit_closed = False + appended = False + if not forked: + if self.closed: + close_conn_reason = ConnectionClosedReason.POOL_CLOSED + elif conn.closed: + # CMAP requires the closed event be emitted after the check in. + emit_closed = True + elif self.stale_generation(conn.generation, conn.service_id): + close_conn_reason = ConnectionClosedReason.STALE + else: + conn.update_last_checkin_time() + conn.update_is_writable(bool(self.is_writable)) + self.conns.appendleft(conn) + appended = True + return close_conn_reason, emit_closed, appended + def checkin(self, conn: Connection) -> None: """Return the connection to the pool, or if it's closed discard it. @@ -1067,44 +1108,49 @@ def checkin(self, conn: Connection) -> None: conn.pinned_txn = False conn.pinned_cursor = False self._pinned_sockets.discard(conn) - with self.lock: - self.active_contexts.discard(conn.cancel_context) + forked = self.pid != os.getpid() + # The pool accounting (requests/active_sockets) must be decremented and + # the connection returned under a single hold of self.size_cond. Under + # gevent, acquiring the lock and Condition.notify() both cooperatively + # yield, so a GreenletExit injected at such a yield -- e.g. a websocket + # handler greenlet being killed while it is checking a connection back + # in after a normal operation -- can interrupt checkin before the + # decrement, leaving requests/active_sockets permanently inflated and + # saturating the size gate (PYTHON-6074). The try/finally below + # re-applies the accounting while unwinding: gevent does not re-throw + # GreenletExit at cooperative yields during unwind, so the re-acquisition + # completes and the accounting is restored. + close_conn_reason: Optional[str] = None + emit_closed = False + accounted = False + try: + with self.size_cond: + close_conn_reason, emit_closed, appended = self._checkin_apply( + conn, txn, cursor, forked + ) + accounted = True + if appended: + # Notify any threads waiting to create a connection. + self._max_connecting_cond.notify() + self.size_cond.notify() + finally: + if not accounted: + with self.size_cond: + close_conn_reason, emit_closed, appended = self._checkin_apply( + conn, txn, cursor, forked + ) + if appended: + self._max_connecting_cond.notify() + self.size_cond.notify() telemetry = self._telemetry if telemetry._should_publish or (telemetry._log and _is_debug_enabled(_CONNECTION_LOGGER)): telemetry.checked_in(conn.id) - if self.pid != os.getpid(): + if emit_closed: + telemetry.connection_closed(conn.id, ConnectionClosedReason.ERROR) + if forked: self.reset_without_pause() - else: - if self.closed: - conn.close_conn(ConnectionClosedReason.POOL_CLOSED) - elif conn.closed: - # CMAP requires the closed event be emitted after the check in. - self._telemetry.connection_closed(conn.id, ConnectionClosedReason.ERROR) - else: - close_conn = False - with self.lock: - # Hold the lock to ensure this section does not race with - # Pool.reset(). - if self.stale_generation(conn.generation, conn.service_id): - close_conn = True - else: - conn.update_last_checkin_time() - conn.update_is_writable(bool(self.is_writable)) - self.conns.appendleft(conn) - # Notify any threads waiting to create a connection. - self._max_connecting_cond.notify() - if close_conn: - conn.close_conn(ConnectionClosedReason.STALE) - - with self.size_cond: - if txn: - self.ntxns -= 1 - elif cursor: - self.ncursors -= 1 - self.requests -= 1 - self.active_sockets -= 1 - self.operation_count -= 1 - self.size_cond.notify() + elif close_conn_reason is not None: + conn.close_conn(close_conn_reason) def _perished(self, conn: Connection) -> bool: """Return True and close the connection if it is "perished". diff --git a/test/asynchronous/test_client.py b/test/asynchronous/test_client.py index c7ddd8b559..5343d6c5b6 100644 --- a/test/asynchronous/test_client.py +++ b/test/asynchronous/test_client.py @@ -2624,6 +2624,104 @@ def timeout_task(): # Unpatch the instance. del pool.connect + @async_client_context.require_sync + def test_gevent_kill_churn_deadlock(self): + """Regression test for PYTHON-6074. + + Under gevent, killing a greenlet that is checking a connection back in + (after a normal operation) interrupts ``Pool.checkin`` during the + contended ``size_cond`` acquisition, before the ``requests``/ + ``active_sockets`` decrement. The accounting stays permanently inflated, + saturating the size gate so every subsequent checkout blocks forever. + This spawns workers that do find_one against a tiny pool plus a reaper + that kills and respawns a worker every few milliseconds, then fails + (rather than hanging) if the op counter stalls while workers are alive. + """ + if not gevent_monkey_patched(): + raise SkipTest("Must be running monkey patched by gevent") + import random + + import gevent + from gevent import Timeout, spawn + + client = self.async_rs_or_single_client(maxPoolSize=2) + coll = client.pymongo_test.coll + coll.insert_one({}) + + op_count = [0] + running = [True] + workers: list = [] + + def worker(): + while running[0]: + try: + coll.find_one({}) + op_count[0] += 1 + time.sleep(0.001) + except Exception: + return + + def reaper(): + while running[0]: + time.sleep(0.003) + if not workers: + continue + idx = random.randrange(len(workers)) + try: + workers[idx].kill(block=False) + except Exception: + pass + workers[idx] = spawn(worker) + + workers[:] = [spawn(worker) for _ in range(8)] + reaper_gr = spawn(reaper) + try: + # Watchdog: fail (never hang) if no op completes for 8s while + # workers are alive. Without the fix the op counter freezes + # permanently once the size gate saturates; with the fix, ops keep + # flowing and the loop simply runs out the deadline. + last = op_count[0] + stale = 0.0 + deadline = time.monotonic() + 12 + while time.monotonic() < deadline: + time.sleep(2) + if op_count[0] == last: + stale += 2 + if stale >= 8: + self.fail( + "Deadlock detected (PYTHON-6074): no ops completed " + f"for {stale:.0f}s, op_count={op_count[0]}" + ) + else: + stale = 0.0 + last = op_count[0] + # Direct liveness probe that does not depend on *when* the leak + # landed: on the fixed driver ops keep flowing, so this completes + # well within the timeout; on the unfixed driver a saturated size + # gate makes it block, the timeout fires and the test fails. + try: + with Timeout(3): + coll.find_one({}) + except Timeout: + self.fail("Pool gate saturated (PYTHON-6074)") + self.assertGreater(op_count[0], 0) + finally: + running[0] = False + gevent.killall(workers, block=False) + try: + reaper_gr.kill(block=False) + except Exception: + pass + # Close the client but never hang on a wedged pool: without the + # PYTHON-6074 fix the pool's size gate is saturated and close() can + # block forever, so bound it. gevent.Timeout is a BaseException, so + # catch BaseException to avoid masking the watchdog's self.fail(). + try: + with Timeout(5): + client.close() + except BaseException: + pass + class TestClientLazyConnect(AsyncIntegrationTest): """Test concurrent operations on a lazily-connecting MongoClient.""" diff --git a/test/test_client.py b/test/test_client.py index 20cceeb037..930d0db940 100644 --- a/test/test_client.py +++ b/test/test_client.py @@ -2575,6 +2575,104 @@ def timeout_task(): # Unpatch the instance. del pool.connect + @client_context.require_sync + def test_gevent_kill_churn_deadlock(self): + """Regression test for PYTHON-6074. + + Under gevent, killing a greenlet that is checking a connection back in + (after a normal operation) interrupts ``Pool.checkin`` during the + contended ``size_cond`` acquisition, before the ``requests``/ + ``active_sockets`` decrement. The accounting stays permanently inflated, + saturating the size gate so every subsequent checkout blocks forever. + This spawns workers that do find_one against a tiny pool plus a reaper + that kills and respawns a worker every few milliseconds, then fails + (rather than hanging) if the op counter stalls while workers are alive. + """ + if not gevent_monkey_patched(): + raise SkipTest("Must be running monkey patched by gevent") + import random + + import gevent + from gevent import Timeout, spawn + + client = self.rs_or_single_client(maxPoolSize=2) + coll = client.pymongo_test.coll + coll.insert_one({}) + + op_count = [0] + running = [True] + workers: list = [] + + def worker(): + while running[0]: + try: + coll.find_one({}) + op_count[0] += 1 + time.sleep(0.001) + except Exception: + return + + def reaper(): + while running[0]: + time.sleep(0.003) + if not workers: + continue + idx = random.randrange(len(workers)) + try: + workers[idx].kill(block=False) + except Exception: + pass + workers[idx] = spawn(worker) + + workers[:] = [spawn(worker) for _ in range(8)] + reaper_gr = spawn(reaper) + try: + # Watchdog: fail (never hang) if no op completes for 8s while + # workers are alive. Without the fix the op counter freezes + # permanently once the size gate saturates; with the fix, ops keep + # flowing and the loop simply runs out the deadline. + last = op_count[0] + stale = 0.0 + deadline = time.monotonic() + 12 + while time.monotonic() < deadline: + time.sleep(2) + if op_count[0] == last: + stale += 2 + if stale >= 8: + self.fail( + "Deadlock detected (PYTHON-6074): no ops completed " + f"for {stale:.0f}s, op_count={op_count[0]}" + ) + else: + stale = 0.0 + last = op_count[0] + # Direct liveness probe that does not depend on *when* the leak + # landed: on the fixed driver ops keep flowing, so this completes + # well within the timeout; on the unfixed driver a saturated size + # gate makes it block, the timeout fires and the test fails. + try: + with Timeout(3): + coll.find_one({}) + except Timeout: + self.fail("Pool gate saturated (PYTHON-6074)") + self.assertGreater(op_count[0], 0) + finally: + running[0] = False + gevent.killall(workers, block=False) + try: + reaper_gr.kill(block=False) + except Exception: + pass + # Close the client but never hang on a wedged pool: without the + # PYTHON-6074 fix the pool's size gate is saturated and close() can + # block forever, so bound it. gevent.Timeout is a BaseException, so + # catch BaseException to avoid masking the watchdog's self.fail(). + try: + with Timeout(5): + client.close() + except BaseException: + pass + class TestClientLazyConnect(IntegrationTest): """Test concurrent operations on a lazily-connecting MongoClient.""" From 7545ad147211e13fce0675d1e2fe54342b37286c Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Fri, 4 Sep 2026 06:46:51 -0500 Subject: [PATCH 2/7] PYTHON-6074 Address review feedback on changelog placement and docstrings --- doc/changelog.rst | 16 ++++++++++++---- pymongo/asynchronous/pool.py | 29 +++++++---------------------- pymongo/synchronous/pool.py | 29 +++++++---------------------- test/asynchronous/test_client.py | 12 +----------- test/test_client.py | 12 +----------- 5 files changed, 28 insertions(+), 70 deletions(-) diff --git a/doc/changelog.rst b/doc/changelog.rst index fa3bad6d5e..04899c96e2 100644 --- a/doc/changelog.rst +++ b/doc/changelog.rst @@ -1,6 +1,18 @@ Changelog ========= +Changes in Version 4.19.0 (2026/XX/XX) +-------------------------------------- + +Bug fixes +......... + +- Fixed a bug where the synchronous client could permanently deadlock under + gevent when a greenlet was killed while checking a connection back into + the pool (`PYTHON-6074`_). + +.. _PYTHON-6074: https://jira.mongodb.org/browse/PYTHON-6074 + Changes in Version 4.18.0 (2026/XX/XX) -------------------------------------- @@ -1043,9 +1055,6 @@ Bug fixes after non-resumable errors (`PYTHON-3389`_). - Fixed a bug where the client could be unable to discover the new primary after a simultaneous replica set election and reconfig (`PYTHON-2970`_). -- Fixed a bug where the synchronous client could permanently deadlock under - gevent when a greenlet was killed while checking a connection back into - the pool (`PYTHON-6074`_). Issues Resolved ............... @@ -1057,7 +1066,6 @@ in this release. .. _PYTHON-2484: https://jira.mongodb.org/browse/PYTHON-2484 .. _PYTHON-2970: https://jira.mongodb.org/browse/PYTHON-2970 .. _PYTHON-3389: https://jira.mongodb.org/browse/PYTHON-3389 -.. _PYTHON-6074: https://jira.mongodb.org/browse/PYTHON-6074 .. _PyMongo 4.3 release notes in JIRA: https://jira.mongodb.org/secure/ReleaseNote.jspa?projectId=10004&version=33425 Changes in Version 4.2.0 (2022/07/20) diff --git a/pymongo/asynchronous/pool.py b/pymongo/asynchronous/pool.py index bb3336b2b3..5d3886815e 100644 --- a/pymongo/asynchronous/pool.py +++ b/pymongo/asynchronous/pool.py @@ -1063,17 +1063,10 @@ async def _get_conn( def _checkin_apply( self, conn: AsyncConnection, txn: bool, cursor: bool, forked: bool ) -> tuple[Optional[str], bool, bool]: - """Apply checkin accounting and decide the connection's disposition. - - Caller must hold ``self.size_cond``. This method performs no cooperative - I/O (no await, no lock/condition acquire, no notify) so it is safe to - call while a gevent greenlet unwinds a ``GreenletExit``: gevent does not - re-throw ``GreenletExit`` at cooperative yields during unwind, so the - re-acquisition in the ``try/finally`` in ``checkin`` completes. The same - guarantee does not hold for an asyncio task cancelled with - ``CancelledError``, which is re-raised at the next await. Returns - ``(close_conn_reason, emit_closed, appended)`` describing work that must - be done outside the lock. See PYTHON-6074. + """Apply checkin accounting; caller holds ``size_cond``. + + No cooperative I/O, so safe while a gevent greenlet unwinds. Returns + ``(close_conn_reason, emit_closed, appended)`` for outside the lock. """ self.active_contexts.discard(conn.cancel_context) if txn: @@ -1113,17 +1106,9 @@ async def checkin(self, conn: AsyncConnection) -> None: conn.pinned_cursor = False self._pinned_sockets.discard(conn) forked = self.pid != os.getpid() - # The pool accounting (requests/active_sockets) must be decremented and - # the connection returned under a single hold of self.size_cond. Under - # gevent, acquiring the lock and Condition.notify() both cooperatively - # yield, so a GreenletExit injected at such a yield -- e.g. a websocket - # handler greenlet being killed while it is checking a connection back - # in after a normal operation -- can interrupt checkin before the - # decrement, leaving requests/active_sockets permanently inflated and - # saturating the size gate (PYTHON-6074). The try/finally below - # re-applies the accounting while unwinding: gevent does not re-throw - # GreenletExit at cooperative yields during unwind, so the re-acquisition - # completes and the accounting is restored. + # Re-apply the accounting if a gevent GreenletExit interrupts during + # the size_cond acquisition; gevent lets the re-acquire complete while + # unwinding (PYTHON-6074). close_conn_reason: Optional[str] = None emit_closed = False accounted = False diff --git a/pymongo/synchronous/pool.py b/pymongo/synchronous/pool.py index 28ff48d0bd..172de271c0 100644 --- a/pymongo/synchronous/pool.py +++ b/pymongo/synchronous/pool.py @@ -1059,17 +1059,10 @@ def _get_conn( def _checkin_apply( self, conn: Connection, txn: bool, cursor: bool, forked: bool ) -> tuple[Optional[str], bool, bool]: - """Apply checkin accounting and decide the connection's disposition. - - Caller must hold ``self.size_cond``. This method performs no cooperative - I/O (no await, no lock/condition acquire, no notify) so it is safe to - call while a gevent greenlet unwinds a ``GreenletExit``: gevent does not - re-throw ``GreenletExit`` at cooperative yields during unwind, so the - re-acquisition in the ``try/finally`` in ``checkin`` completes. The same - guarantee does not hold for an asyncio task cancelled with - ``CancelledError``, which is re-raised at the next await. Returns - ``(close_conn_reason, emit_closed, appended)`` describing work that must - be done outside the lock. See PYTHON-6074. + """Apply checkin accounting; caller holds ``size_cond``. + + No cooperative I/O, so safe while a gevent greenlet unwinds. Returns + ``(close_conn_reason, emit_closed, appended)`` for outside the lock. """ self.active_contexts.discard(conn.cancel_context) if txn: @@ -1109,17 +1102,9 @@ def checkin(self, conn: Connection) -> None: conn.pinned_cursor = False self._pinned_sockets.discard(conn) forked = self.pid != os.getpid() - # The pool accounting (requests/active_sockets) must be decremented and - # the connection returned under a single hold of self.size_cond. Under - # gevent, acquiring the lock and Condition.notify() both cooperatively - # yield, so a GreenletExit injected at such a yield -- e.g. a websocket - # handler greenlet being killed while it is checking a connection back - # in after a normal operation -- can interrupt checkin before the - # decrement, leaving requests/active_sockets permanently inflated and - # saturating the size gate (PYTHON-6074). The try/finally below - # re-applies the accounting while unwinding: gevent does not re-throw - # GreenletExit at cooperative yields during unwind, so the re-acquisition - # completes and the accounting is restored. + # Re-apply the accounting if a gevent GreenletExit interrupts during + # the size_cond acquisition; gevent lets the re-acquire complete while + # unwinding (PYTHON-6074). close_conn_reason: Optional[str] = None emit_closed = False accounted = False diff --git a/test/asynchronous/test_client.py b/test/asynchronous/test_client.py index 5343d6c5b6..71d2d2f2a9 100644 --- a/test/asynchronous/test_client.py +++ b/test/asynchronous/test_client.py @@ -2626,17 +2626,7 @@ def timeout_task(): @async_client_context.require_sync def test_gevent_kill_churn_deadlock(self): - """Regression test for PYTHON-6074. - - Under gevent, killing a greenlet that is checking a connection back in - (after a normal operation) interrupts ``Pool.checkin`` during the - contended ``size_cond`` acquisition, before the ``requests``/ - ``active_sockets`` decrement. The accounting stays permanently inflated, - saturating the size gate so every subsequent checkout blocks forever. - This spawns workers that do find_one against a tiny pool plus a reaper - that kills and respawns a worker every few milliseconds, then fails - (rather than hanging) if the op counter stalls while workers are alive. - """ + """A greenlet killed mid-operation must not leave the pool deadlocked (PYTHON-6074).""" if not gevent_monkey_patched(): raise SkipTest("Must be running monkey patched by gevent") import random diff --git a/test/test_client.py b/test/test_client.py index 930d0db940..7e50e9e67c 100644 --- a/test/test_client.py +++ b/test/test_client.py @@ -2577,17 +2577,7 @@ def timeout_task(): @client_context.require_sync def test_gevent_kill_churn_deadlock(self): - """Regression test for PYTHON-6074. - - Under gevent, killing a greenlet that is checking a connection back in - (after a normal operation) interrupts ``Pool.checkin`` during the - contended ``size_cond`` acquisition, before the ``requests``/ - ``active_sockets`` decrement. The accounting stays permanently inflated, - saturating the size gate so every subsequent checkout blocks forever. - This spawns workers that do find_one against a tiny pool plus a reaper - that kills and respawns a worker every few milliseconds, then fails - (rather than hanging) if the op counter stalls while workers are alive. - """ + """A greenlet killed mid-operation must not leave the pool deadlocked (PYTHON-6074).""" if not gevent_monkey_patched(): raise SkipTest("Must be running monkey patched by gevent") import random From bd1ec24b979f5ceabf4ffe2edbfc07dca4034030 Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Fri, 4 Sep 2026 08:24:54 -0500 Subject: [PATCH 3/7] PYTHON-6074 Catch gevent Timeout explicitly in kill-churn test cleanup --- test/asynchronous/test_client.py | 6 +++--- test/test_client.py | 6 +++--- 2 files changed, 6 insertions(+), 6 deletions(-) diff --git a/test/asynchronous/test_client.py b/test/asynchronous/test_client.py index 71d2d2f2a9..458ee4e503 100644 --- a/test/asynchronous/test_client.py +++ b/test/asynchronous/test_client.py @@ -2704,12 +2704,12 @@ def reaper(): pass # Close the client but never hang on a wedged pool: without the # PYTHON-6074 fix the pool's size gate is saturated and close() can - # block forever, so bound it. gevent.Timeout is a BaseException, so - # catch BaseException to avoid masking the watchdog's self.fail(). + # block forever, so bound it and let the watchdog's self.fail() + # propagate. try: with Timeout(5): client.close() - except BaseException: + except Timeout: pass diff --git a/test/test_client.py b/test/test_client.py index 7e50e9e67c..eaf9351311 100644 --- a/test/test_client.py +++ b/test/test_client.py @@ -2655,12 +2655,12 @@ def reaper(): pass # Close the client but never hang on a wedged pool: without the # PYTHON-6074 fix the pool's size gate is saturated and close() can - # block forever, so bound it. gevent.Timeout is a BaseException, so - # catch BaseException to avoid masking the watchdog's self.fail(). + # block forever, so bound it and let the watchdog's self.fail() + # propagate. try: with Timeout(5): client.close() - except BaseException: + except Timeout: pass From 249605f3ca85548d03a97dd98ac492be57240c87 Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Fri, 4 Sep 2026 14:06:39 -0500 Subject: [PATCH 4/7] PYTHON-6074 Add AMPLIFY_RACE synthetic amplification to kill-churn test --- test/asynchronous/test_client.py | 28 ++++++++++++++++++++++++++++ test/test_client.py | 28 ++++++++++++++++++++++++++++ 2 files changed, 56 insertions(+) diff --git a/test/asynchronous/test_client.py b/test/asynchronous/test_client.py index 458ee4e503..2d812d221b 100644 --- a/test/asynchronous/test_client.py +++ b/test/asynchronous/test_client.py @@ -2632,8 +2632,31 @@ def test_gevent_kill_churn_deadlock(self): import random import gevent + import gevent.thread as _gthread from gevent import Timeout, spawn + # PYTHON-6074: widen gevent's courtesy-yield window (the bare + # sleep() on a failed non-blocking lock acquire, which + # Condition.notify()'s _is_owned() hits on every checkin) so a + # GreenletExit lands in that window deterministically. Synthetic: + # only the bare sleep() is widened; sleep(t) with args passes + # through, so the reaper and watchdog timings are unchanged. Set + # AMPLIFY_RACE=1 to enable; the unfixed test then deadlocks + # within seconds every run. + if os.environ.get("AMPLIFY_RACE", "0") == "1": + _AMPLIFY_SECONDS = float(os.environ.get("AMPLIFY_SECONDS", "0.02")) + + _orig_thread_sleep = _gthread.sleep + + def _amplified_sleep(*args): + if not args: # bare sleep(): the courtesy yield on a failed + # non-blocking acquire (Condition.notify -> _is_owned -> acquire(False)) + _orig_thread_sleep(_AMPLIFY_SECONDS) + else: # sleep(0.001), sleep(2), etc.: passthrough + _orig_thread_sleep(*args) + + _gthread.sleep = _amplified_sleep + client = self.async_rs_or_single_client(maxPoolSize=2) coll = client.pymongo_test.coll coll.insert_one({}) @@ -2694,6 +2717,11 @@ def reaper(): coll.find_one({}) except Timeout: self.fail("Pool gate saturated (PYTHON-6074)") + # Deterministic check: a saturated size gate pins the pool's + # checkout counters at maxPoolSize (PYTHON-6074). + pool = async_get_pool(client) # type:ignore + self.assertLess(pool.requests, pool.max_pool_size) + self.assertLess(pool.active_sockets, pool.max_pool_size) self.assertGreater(op_count[0], 0) finally: running[0] = False diff --git a/test/test_client.py b/test/test_client.py index eaf9351311..f7d5a1895f 100644 --- a/test/test_client.py +++ b/test/test_client.py @@ -2583,8 +2583,31 @@ def test_gevent_kill_churn_deadlock(self): import random import gevent + import gevent.thread as _gthread from gevent import Timeout, spawn + # PYTHON-6074: widen gevent's courtesy-yield window (the bare + # sleep() on a failed non-blocking lock acquire, which + # Condition.notify()'s _is_owned() hits on every checkin) so a + # GreenletExit lands in that window deterministically. Synthetic: + # only the bare sleep() is widened; sleep(t) with args passes + # through, so the reaper and watchdog timings are unchanged. Set + # AMPLIFY_RACE=1 to enable; the unfixed test then deadlocks + # within seconds every run. + if os.environ.get("AMPLIFY_RACE", "0") == "1": + _AMPLIFY_SECONDS = float(os.environ.get("AMPLIFY_SECONDS", "0.02")) + + _orig_thread_sleep = _gthread.sleep + + def _amplified_sleep(*args): + if not args: # bare sleep(): the courtesy yield on a failed + # non-blocking acquire (Condition.notify -> _is_owned -> acquire(False)) + _orig_thread_sleep(_AMPLIFY_SECONDS) + else: # sleep(0.001), sleep(2), etc.: passthrough + _orig_thread_sleep(*args) + + _gthread.sleep = _amplified_sleep + client = self.rs_or_single_client(maxPoolSize=2) coll = client.pymongo_test.coll coll.insert_one({}) @@ -2645,6 +2668,11 @@ def reaper(): coll.find_one({}) except Timeout: self.fail("Pool gate saturated (PYTHON-6074)") + # Deterministic check: a saturated size gate pins the pool's + # checkout counters at maxPoolSize (PYTHON-6074). + pool = get_pool(client) # type:ignore + self.assertLess(pool.requests, pool.max_pool_size) + self.assertLess(pool.active_sockets, pool.max_pool_size) self.assertGreater(op_count[0], 0) finally: running[0] = False From b41c933a8b49b51ca97baf58a496f7df708266f0 Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Fri, 4 Sep 2026 17:16:59 -0500 Subject: [PATCH 5/7] PYTHON-6074 Set accounted before notify in checkout error handler Under gevent, notify() is a yield point, so a kill delivered inside it left accounted False and the fallback decremented the accounting a second time. Set the flag before notify() so only interruption during condition acquisition triggers the fallback. --- pymongo/asynchronous/pool.py | 23 ++++++++++++++++++----- pymongo/synchronous/pool.py | 23 ++++++++++++++++++----- test/asynchronous/test_pooling.py | 30 ++++++++++++++++++++++++++++++ test/test_pooling.py | 30 ++++++++++++++++++++++++++++++ 4 files changed, 96 insertions(+), 10 deletions(-) diff --git a/pymongo/asynchronous/pool.py b/pymongo/asynchronous/pool.py index 5d3886815e..434af0ef5a 100644 --- a/pymongo/asynchronous/pool.py +++ b/pymongo/asynchronous/pool.py @@ -1044,11 +1044,24 @@ async def _get_conn( if conn: # We checked out a socket but authentication failed. await conn.close_conn(ConnectionClosedReason.ERROR) - async with self.size_cond: - self.requests -= 1 - if incremented: - self.active_sockets -= 1 - self.size_cond.notify() + # Re-apply the accounting if a GreenletExit interrupts + # during the size_cond acquisition; during unwind gevent + # lets the re-acquire complete (PYTHON-6074). + accounted = False + try: + async with self.size_cond: + self.requests -= 1 + if incremented: + self.active_sockets -= 1 + accounted = True + self.size_cond.notify() + finally: + if not accounted: + async with self.size_cond: + self.requests -= 1 + if incremented: + self.active_sockets -= 1 + self.size_cond.notify() if not emitted_event: self._telemetry.checkout_failed( diff --git a/pymongo/synchronous/pool.py b/pymongo/synchronous/pool.py index 172de271c0..255dd09bf0 100644 --- a/pymongo/synchronous/pool.py +++ b/pymongo/synchronous/pool.py @@ -1040,11 +1040,24 @@ def _get_conn( if conn: # We checked out a socket but authentication failed. conn.close_conn(ConnectionClosedReason.ERROR) - with self.size_cond: - self.requests -= 1 - if incremented: - self.active_sockets -= 1 - self.size_cond.notify() + # Re-apply the accounting if a GreenletExit interrupts + # during the size_cond acquisition; during unwind gevent + # lets the re-acquire complete (PYTHON-6074). + accounted = False + try: + with self.size_cond: + self.requests -= 1 + if incremented: + self.active_sockets -= 1 + accounted = True + self.size_cond.notify() + finally: + if not accounted: + with self.size_cond: + self.requests -= 1 + if incremented: + self.active_sockets -= 1 + self.size_cond.notify() if not emitted_event: self._telemetry.checkout_failed( diff --git a/test/asynchronous/test_pooling.py b/test/asynchronous/test_pooling.py index ef670774c8..9f94c90eea 100644 --- a/test/asynchronous/test_pooling.py +++ b/test/asynchronous/test_pooling.py @@ -277,6 +277,36 @@ def add(self, item): self.assertEqual(0, cx_pool.active_sockets) self.assertEqual(0, cx_pool.requests) + async def test_checkout_error_accounting_no_double_decrement(self): + # PYTHON-6074: an exception delivered while the checkout error handler + # is inside size_cond.notify() (a yield point under gevent, where a + # greenlet can be killed) must not cause the accounting to be applied + # a second time by the handler's fallback. + cx_pool = await self.create_pool(max_pool_size=1) + + real_notify = cx_pool.size_cond.notify + notify_calls = [] + + def notify(): + notify_calls.append(1) + if len(notify_calls) == 1: + # Simulate a kill delivered at the notify() yield point. + raise KeyboardInterrupt() + real_notify() + + cx_pool.size_cond.notify = notify + try: + with patch.object(cx_pool, "connect", side_effect=asyncio.CancelledError()): + with self.assertRaises(KeyboardInterrupt): + async with cx_pool.checkout(): + pass + finally: + cx_pool.size_cond.notify = real_notify + + # Accounting was applied exactly once. + self.assertEqual(0, cx_pool.requests) + self.assertEqual(0, cx_pool.active_sockets) + async def test_pool_removes_closed_socket(self): # Test that Pool removes explicitly closed socket. cx_pool = await self.create_pool() diff --git a/test/test_pooling.py b/test/test_pooling.py index 64e6aa931f..f241cc0cbb 100644 --- a/test/test_pooling.py +++ b/test/test_pooling.py @@ -277,6 +277,36 @@ def add(self, item): self.assertEqual(0, cx_pool.active_sockets) self.assertEqual(0, cx_pool.requests) + def test_checkout_error_accounting_no_double_decrement(self): + # PYTHON-6074: an exception delivered while the checkout error handler + # is inside size_cond.notify() (a yield point under gevent, where a + # greenlet can be killed) must not cause the accounting to be applied + # a second time by the handler's fallback. + cx_pool = self.create_pool(max_pool_size=1) + + real_notify = cx_pool.size_cond.notify + notify_calls = [] + + def notify(): + notify_calls.append(1) + if len(notify_calls) == 1: + # Simulate a kill delivered at the notify() yield point. + raise KeyboardInterrupt() + real_notify() + + cx_pool.size_cond.notify = notify + try: + with patch.object(cx_pool, "connect", side_effect=asyncio.CancelledError()): + with self.assertRaises(KeyboardInterrupt): + with cx_pool.checkout(): + pass + finally: + cx_pool.size_cond.notify = real_notify + + # Accounting was applied exactly once. + self.assertEqual(0, cx_pool.requests) + self.assertEqual(0, cx_pool.active_sockets) + def test_pool_removes_closed_socket(self): # Test that Pool removes explicitly closed socket. cx_pool = self.create_pool() From f877d7a8eeda6ceb241e8565b05a3fd1917ba388 Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Fri, 4 Sep 2026 19:00:47 -0500 Subject: [PATCH 6/7] PYTHON-6074 Cover checkout error handler fallback on acquire Adds a test interrupting the error handler while it waits to acquire size_cond, covering the fallback that re-applies the checkout accounting. --- test/asynchronous/test_pooling.py | 35 +++++++++++++++++++++++++++++++ test/test_pooling.py | 35 +++++++++++++++++++++++++++++++ 2 files changed, 70 insertions(+) diff --git a/test/asynchronous/test_pooling.py b/test/asynchronous/test_pooling.py index 9f94c90eea..661bd4e3d2 100644 --- a/test/asynchronous/test_pooling.py +++ b/test/asynchronous/test_pooling.py @@ -307,6 +307,41 @@ def notify(): self.assertEqual(0, cx_pool.requests) self.assertEqual(0, cx_pool.active_sockets) + async def test_checkout_error_accounting_on_kill_during_acquire(self): + # PYTHON-6074: an exception delivered while the checkout error + # handler is waiting to acquire size_cond (a yield point under + # gevent) must not leak the checkout accounting; the handler's + # fallback re-applies it. + cx_pool = await self.create_pool(max_pool_size=1) + + class _InterruptOnSecondEnter(type(cx_pool.size_cond)): + def __init__(self, lock): + super().__init__(lock) + self.enters = 0 + + async def __aenter__(self): + self.enters += 1 + if self.enters == 2: + # First enter is the checkout semaphore, second is the + # error handler. Simulate a kill delivered while blocked + # on the second. + raise KeyboardInterrupt() + return await super().__aenter__() + + async def __aexit__(self, *args): + return await super().__aexit__(*args) + + cx_pool.size_cond = _InterruptOnSecondEnter(cx_pool.size_cond._lock) + + with patch.object(cx_pool, "connect", side_effect=asyncio.CancelledError()): + with self.assertRaises(KeyboardInterrupt): + async with cx_pool.checkout(): + pass + + # The fallback applied the accounting exactly once. + self.assertEqual(0, cx_pool.requests) + self.assertEqual(0, cx_pool.active_sockets) + async def test_pool_removes_closed_socket(self): # Test that Pool removes explicitly closed socket. cx_pool = await self.create_pool() diff --git a/test/test_pooling.py b/test/test_pooling.py index f241cc0cbb..a3f0eaf589 100644 --- a/test/test_pooling.py +++ b/test/test_pooling.py @@ -307,6 +307,41 @@ def notify(): self.assertEqual(0, cx_pool.requests) self.assertEqual(0, cx_pool.active_sockets) + def test_checkout_error_accounting_on_kill_during_acquire(self): + # PYTHON-6074: an exception delivered while the checkout error + # handler is waiting to acquire size_cond (a yield point under + # gevent) must not leak the checkout accounting; the handler's + # fallback re-applies it. + cx_pool = self.create_pool(max_pool_size=1) + + class _InterruptOnSecondEnter(type(cx_pool.size_cond)): + def __init__(self, lock): + super().__init__(lock) + self.enters = 0 + + def __enter__(self): + self.enters += 1 + if self.enters == 2: + # First enter is the checkout semaphore, second is the + # error handler. Simulate a kill delivered while blocked + # on the second. + raise KeyboardInterrupt() + return super().__enter__() + + def __exit__(self, *args): + return super().__exit__(*args) + + cx_pool.size_cond = _InterruptOnSecondEnter(cx_pool.size_cond._lock) + + with patch.object(cx_pool, "connect", side_effect=asyncio.CancelledError()): + with self.assertRaises(KeyboardInterrupt): + with cx_pool.checkout(): + pass + + # The fallback applied the accounting exactly once. + self.assertEqual(0, cx_pool.requests) + self.assertEqual(0, cx_pool.active_sockets) + def test_pool_removes_closed_socket(self): # Test that Pool removes explicitly closed socket. cx_pool = self.create_pool() From 1e77b06d7ec967187065bad0e3448fdfb4fc16b2 Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Fri, 4 Sep 2026 19:45:35 -0500 Subject: [PATCH 7/7] PYTHON-6074 Restore gevent sleep monkeypatch after kill-churn test When AMPLIFY_RACE is enabled the test reassigns gevent.thread.sleep globally; register a cleanup so the widened sleep window doesn't leak into later tests. --- test/asynchronous/test_client.py | 1 + test/test_client.py | 1 + 2 files changed, 2 insertions(+) diff --git a/test/asynchronous/test_client.py b/test/asynchronous/test_client.py index 2d812d221b..b1dbd461be 100644 --- a/test/asynchronous/test_client.py +++ b/test/asynchronous/test_client.py @@ -2656,6 +2656,7 @@ def _amplified_sleep(*args): _orig_thread_sleep(*args) _gthread.sleep = _amplified_sleep + self.addCleanup(setattr, _gthread, "sleep", _orig_thread_sleep) client = self.async_rs_or_single_client(maxPoolSize=2) coll = client.pymongo_test.coll diff --git a/test/test_client.py b/test/test_client.py index f7d5a1895f..39fec44f82 100644 --- a/test/test_client.py +++ b/test/test_client.py @@ -2607,6 +2607,7 @@ def _amplified_sleep(*args): _orig_thread_sleep(*args) _gthread.sleep = _amplified_sleep + self.addCleanup(setattr, _gthread, "sleep", _orig_thread_sleep) client = self.rs_or_single_client(maxPoolSize=2) coll = client.pymongo_test.coll