From fb4dc18b1321055fe970ab4bd585e2b45d93b975 Mon Sep 17 00:00:00 2001 From: Yaniv Kaul Date: Fri, 14 Aug 2026 23:53:23 +0300 Subject: [PATCH] Fix cross-lock race in libev reactor thread exit check (free-threaded 3.14t hang) LibevLoop._live_conns is written under a lock in connection_created()/connection_destroyed(), while _run_loop()'s exit check decides whether to stop the reactor thread. An earlier version of this fix moved the _live_conns read onto the same lock the writers used, but the started/shutdown flags were still read and set under a separate lock afterward -- leaving a gap where connection_created() could still register a connection between the read and the state transition. CI on that version reproduced the exact hang this fix is meant to eliminate. Fix: merge _lock and the former _conn_set_lock into a single lock that guards both _live_conns/_new_conns/_closed_conns *and* the _started/_shutdown transitions read in the exit check. This makes the exit decision and connection registration mutually exclusive rather than just reading from a shared lock: a concurrent connection_created() either finishes before the exit check's critical section (its connection is visible in _live_conns, so the reactor keeps running) or finishes after _started is set to False inside that same critical section (so the subsequent maybe_start() call, which always follows connection_created(), sees _started == False and starts a fresh thread). There is no interleaving in which the new connection is invisible to both checks, closing the race rather than narrowing it. The two locks didn't need to stay separate: _run_loop() already nested "with self._lock: with self._conn_set_lock:", and connection_created()/connection_destroyed()/_loop_will_run() never call anything that reacquires _lock, so merging them introduces no reentrancy or ordering issue. Also add a regression test (LibevLoopRaceTest) that forces the exact interleaving from issue #980 deterministically: it pauses the reactor thread's exit-check via an instrumented lock right after it starts deciding, then tries to register a connection from another thread. The test asserts connection_created() cannot complete until the reactor's decision is committed, and that maybe_start() correctly restarts the reactor if the connection lands just after. Verified this test fails (reliably, not flakily) against the git history's prior attempt at this fix and passes against this one. Fixes #980. --- cassandra/io/libevreactor.py | 30 ++++- tests/unit/io/test_libevreactor.py | 176 ++++++++++++++++++++++++++++- 2 files changed, 199 insertions(+), 7 deletions(-) diff --git a/cassandra/io/libevreactor.py b/cassandra/io/libevreactor.py index 6cceb6c6bc..8e125e35fc 100644 --- a/cassandra/io/libevreactor.py +++ b/cassandra/io/libevreactor.py @@ -57,19 +57,27 @@ def __init__(self): self._started = False self._shutdown = False + # Single lock for _started/_shutdown *and* _live_conns/_new_conns/ + # _closed_conns. The exit check in _run_loop() must see the started/ + # shutdown flags and the live connection set as one atomic snapshot, + # otherwise connection_created() can register a connection in the + # instant after the exit check reads an empty _live_conns but before + # _started is flipped to False -- a lost wakeup that hangs the + # connection forever (see issue #980). Two separate locks can't give + # that atomicity no matter which one each side takes, so there is + # only one lock here, not a hold-both-locks protocol. self._lock = Lock() self._lock_thread = Lock() self._thread = None # set of all connections; only replaced with a new copy - # while holding _conn_set_lock, never modified in place + # while holding _lock, never modified in place self._live_conns = set() # newly created connections that need their write/read watcher started self._new_conns = set() # recently closed connections that need their write/read watcher stopped self._closed_conns = set() - self._conn_set_lock = Lock() self._preparer = libev.Prepare(self._loop, self._loop_will_run) # prevent _preparer from keeping the loop from returning @@ -101,6 +109,16 @@ def _run_loop(self): self._loop.start() # there are still active watchers, no deadlock with self._lock: + # Reading _live_conns and deciding/committing the exit here + # happen atomically under the same lock that guards + # connection_created()/connection_destroyed(). So any + # concurrent connection_created() either finishes-before this + # read (its connection is seen in _live_conns, loop + # restarts) or finishes-after this block sets _started = + # False (maybe_start(), called right after + # connection_created(), then observes _started == False and + # starts a fresh thread). There is no interleaving in which + # the new connection is invisible to both. if not self._shutdown and self._live_conns: log.debug("Restarting event loop") continue @@ -159,7 +177,7 @@ def notify(self): self._notifier.send() def connection_created(self, conn): - with self._conn_set_lock: + with self._lock: new_live_conns = self._live_conns.copy() new_live_conns.add(conn) self._live_conns = new_live_conns @@ -169,7 +187,7 @@ def connection_created(self, conn): self._new_conns = new_new_conns def connection_destroyed(self, conn): - with self._conn_set_lock: + with self._lock: new_conns = self._new_conns.copy() new_conns.discard(conn) self._new_conns = new_conns @@ -198,7 +216,7 @@ def _loop_will_run(self, prepare): changed = True if self._new_conns: - with self._conn_set_lock: + with self._lock: to_start = self._new_conns self._new_conns = set() @@ -209,7 +227,7 @@ def _loop_will_run(self, prepare): changed = True if self._closed_conns: - with self._conn_set_lock: + with self._lock: to_stop = self._closed_conns self._closed_conns = set() diff --git a/tests/unit/io/test_libevreactor.py b/tests/unit/io/test_libevreactor.py index 000930fd43..13ed2e0195 100644 --- a/tests/unit/io/test_libevreactor.py +++ b/tests/unit/io/test_libevreactor.py @@ -11,6 +11,7 @@ # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. +import threading import unittest from unittest.mock import patch, Mock @@ -20,9 +21,10 @@ try: from cassandra.io.libevreactor import _cleanup as libev__cleanup - from cassandra.io.libevreactor import LibevConnection + from cassandra.io.libevreactor import LibevConnection, LibevLoop except (ImportError, DependencyException): LibevConnection = None # noqa + LibevLoop = None # noqa from tests.unit.io.utils import ReactorTestMixin, TimerTestMixin @@ -95,6 +97,178 @@ def test_watchers_are_finished(self): _global_loop._preparer.start() +class _InstrumentedLock(object): + """ + Wraps a real threading.Lock and calls a hook the first time it is + acquired. Used to pause a thread *while it holds the lock* so a second + thread's attempt to acquire the same lock can be observed as blocking + (or not). + """ + + def __init__(self, on_first_acquire): + self._real_lock = threading.Lock() + self._on_first_acquire = on_first_acquire + self._acquire_count = 0 + self._count_lock = threading.Lock() + + def acquire(self, *args, **kwargs): + got = self._real_lock.acquire(*args, **kwargs) + if got: + with self._count_lock: + self._acquire_count += 1 + first = self._acquire_count == 1 + if first: + self._on_first_acquire() + return got + + def release(self): + self._real_lock.release() + + def __enter__(self): + self.acquire() + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + self.release() + + +class LibevLoopRaceTest(unittest.TestCase): + """ + Regression tests for GH-980: LibevLoop._run_loop()'s decision to exit + the reactor thread (based on _live_conns being empty) must be atomic + with connection_created() registering a new connection. If it isn't, + a connection can be added in the instant after the exit check reads an + empty _live_conns but before the reactor commits to exiting -- and + since maybe_start() (called right after connection_created()) also + sees the stale "already started" state, nobody ever starts a new + reactor thread for that connection. It is silently orphaned forever. + """ + + def setUp(self): + if LibevLoop is None: + raise unittest.SkipTest('libev does not appear to be installed correctly') + + def test_connection_created_cannot_race_the_exit_check(self): + """ + Force the exact interleaving that produces the hang: pause the + reactor thread right after it enters the critical section that + decides whether to exit (i.e. right after it acquires the lock + that must guard both _live_conns and the started/shutdown state), + then try to register a new connection from another thread. With + the fix, connection_created() must block until the reactor + finishes its decision, so the two operations can never interleave. + + @jira_ticket GH-980 + """ + loop = LibevLoop() + + # No real watchers are involved in this test; make each pass of + # the reactor loop return immediately. + loop._loop = Mock() + loop._shutdown = False + loop._live_conns = set() # nothing live -> the reactor wants to exit + loop._started = True # simulate an already-running reactor thread + + reactor_in_critical_section = threading.Event() + release_reactor = threading.Event() + + def pause_reactor(): + reactor_in_critical_section.set() + # Hold the lock open long enough to give connection_created() + # a real chance to race in while we're "deciding". + release_reactor.wait(timeout=5) + + loop._lock = _InstrumentedLock(pause_reactor) + + reactor_thread = threading.Thread(target=loop._run_loop, name="test_reactor", daemon=True) + reactor_thread.start() + self.addCleanup(reactor_thread.join, 5) + + self.assertTrue( + reactor_in_critical_section.wait(timeout=5), + "reactor thread never entered its exit-check critical section") + + conn = Mock() + connection_created_done = threading.Event() + + def create_connection(): + loop.connection_created(conn) + connection_created_done.set() + + creator_thread = threading.Thread(target=create_connection, name="test_creator", daemon=True) + creator_thread.start() + self.addCleanup(creator_thread.join, 5) + + # While the reactor is still deciding, connection_created() must + # NOT be able to complete -- if it does, the exit decision and the + # connection registration were not atomic (the bug from GH-980: + # a two-lock split where a writer could slip a connection in + # between the reactor's read of _live_conns and its commit to + # exit/started=False). + raced_in = connection_created_done.wait(timeout=0.5) + self.assertFalse( + raced_in, + "connection_created() completed while the reactor thread was " + "still deciding whether to exit -- the exit check and " + "connection registration are not atomic, reproducing GH-980") + + # Let the reactor finish its decision (it will see the pre-race + # empty _live_conns, and exit). + release_reactor.set() + creator_thread.join(timeout=5) + reactor_thread.join(timeout=5) + + self.assertFalse(reactor_thread.is_alive()) + self.assertTrue(connection_created_done.is_set()) + # The connection was registered (never lost)... + self.assertIn(conn, loop._live_conns) + # ...and because it landed strictly after the reactor committed to + # exiting, _started correctly reflects "not running": a + # subsequent maybe_start() (as LibevConnection.__init__ always + # calls right after connection_created()) will see this and spin + # up a fresh thread instead of stranding the connection. + self.assertFalse(loop._started) + + with patch('cassandra.io.libevreactor.Thread') as mock_thread_cls: + loop.maybe_start() + mock_thread_cls.assert_called_once() + self.assertTrue(loop._started) + + def test_live_connection_prevents_exit(self): + """ + Sanity check for the other side of the same critical section: if + connection_created() completes (and is visible) before the + reactor's exit check runs, the reactor must see the connection and + keep the loop running rather than exit. + """ + loop = LibevLoop() + + conn = Mock() + loop.connection_created(conn) + loop._shutdown = False + + calls = {'n': 0} + + def fake_start(): + calls['n'] += 1 + if calls['n'] == 1: + return + # Second pass: simulate the connection being closed so the + # loop can actually terminate instead of spinning forever. + loop.connection_destroyed(conn) + + loop._loop = Mock() + loop._loop.start = fake_start + + reactor_thread = threading.Thread(target=loop._run_loop, name="test_reactor", daemon=True) + reactor_thread.start() + reactor_thread.join(timeout=5) + + self.assertFalse(reactor_thread.is_alive()) + self.assertEqual(calls['n'], 2) + self.assertFalse(loop._started) + + class LibevTimerPatcher(unittest.TestCase): @classmethod