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