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.

Author: mykaul

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

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