Merge pull request datastax#1048 from nschrader/python-1044

[Python 1044] Driver hangs/deadlock if all connections dropped by heartbeat whilst request in flight and request times out

Author: aboudreault

CHANGELOG.rst

@@ -5,6 +5,7 @@ Unreleased
 Bug Fixes
 ---------
 * Connection setup methods prevent using ExecutionProfile in cqlengine (PYTHON-1009)
+* Driver deadlock if all connections dropped by heartbeat whilst request in flight and request times out (PYTHON-1044)
 
 3.19.0
 ======

cassandra/cluster.py

@@ -3722,10 +3722,14 @@ def _on_timeout(self, _attempts=0):
         if self._connection is not None:
             try:
                 self._connection._requests.pop(self._req_id)
-            # This prevents the race condition of the
-            # event loop thread just receiving the waited message
-            # If it arrives after this, it will be ignored
+            # PYTHON-1044
+            # This request might have been removed from the connection after the latter was defunct by heartbeat.
+            # We should still raise OperationTimedOut to reject the future so that the main event thread will not
+            # wait for it endlessly
             except KeyError:
+                key = "Connection defunct by heartbeat"
+                errors = {key: "Client request timeout. See Session.execute[_async](timeout)"}
+                self._set_final_exception(OperationTimedOut(errors, self._current_host))
                 return
 
             pool = self.session._pools.get(self._current_host)

tests/integration/__init__.py

@@ -126,7 +126,7 @@ def _get_dse_version_from_cass(cass_version):
 CASSANDRA_IP = os.getenv('CASSANDRA_IP', '127.0.0.1')
 CASSANDRA_DIR = os.getenv('CASSANDRA_DIR', None)
 
-default_cassandra_version = '3.11'
+default_cassandra_version = '3.11.4'
 cv_string = os.getenv('CASSANDRA_VERSION', default_cassandra_version)
 mcv_string = os.getenv('MAPPED_CASSANDRA_VERSION', None)
 try:
@@ -366,6 +366,9 @@ def use_cluster(cluster_name, nodes, ipformat=None, start=True, workloads=None,
     elif ccm_options is None:
         ccm_options = CCM_KWARGS.copy()
 
+    if 'version' in ccm_options and not isinstance(ccm_options['version'], Version):
+        ccm_options['version'] = Version(ccm_options['version'])
+
     cassandra_version = ccm_options.get('version', CCM_VERSION)
     dse_version = ccm_options.get('version', DSE_VERSION)
 

tests/integration/simulacron/test_connection.py

@@ -27,8 +27,9 @@
 from cassandra.policies import HostStateListener, RoundRobinPolicy
 from cassandra.io.asyncorereactor import AsyncoreConnection
 from tests import connection_class, thread_pool_executor_class
+from tests.unit.cython.utils import cythontest
 from tests.integration import (PROTOCOL_VERSION, requiressimulacron)
-from tests.integration.util import assert_quiescent_pool_state
+from tests.integration.util import assert_quiescent_pool_state, late
 from tests.integration.simulacron import SimulacronBase
 from tests.integration.simulacron.utils import (NO_THEN, PrimeOptions,
                                                 prime_query, prime_request,
@@ -178,6 +179,80 @@ def test_callbacks_and_pool_when_oto(self):
         errback.assert_called_once()
         callback.assert_not_called()
 
+    @cythontest
+    def test_heartbeat_defunct_deadlock(self):
+        """
+        Ensure that there is no deadlock when request is in-flight and heartbeat defuncts connection
+        @since 3.16
+        @jira_ticket PYTHON-1044
+        @expected_result an OperationTimeout is raised and no deadlock occurs
+
+        @test_category connection
+        """
+        start_and_prime_singledc()
+
+        # This is all about timing. We will need the QUERY response future to time out and the heartbeat to defunct
+        # at the same moment. The latter will schedule a QUERY retry to another node in case the pool is not
+        # already shut down.  If and only if the response future timeout falls in between the retry scheduling and
+        # its execution the deadlock occurs. The odds are low, so we need to help fate a bit:
+        # 1) Make one heartbeat messages be sent to every node
+        # 2) Our QUERY goes always to the same host
+        # 3) This host needs to defunct first
+        # 4) Open a small time window for the response future timeout, i.e. block executor threads for retry
+        #    execution and last connection to defunct
+        query_to_prime = "SELECT * from testkesypace.testtable"
+        query_host = "127.0.0.2"
+        heartbeat_interval = 1
+        heartbeat_timeout = 1
+        lag = 0.05
+        never = 9999
+
+        class PatchedRoundRobinPolicy(RoundRobinPolicy):
+            # Send always to same host
+            def make_query_plan(self, working_keyspace=None, query=None):
+                print query
+                print self._live_hosts
+                if query and query.query_string == query_to_prime:
+                    return filter(lambda h: h == query_host, self._live_hosts)
+                else:
+                    return super(PatchedRoundRobinPolicy, self).make_query_plan()
+
+        class PatchedCluster(Cluster):
+            # Make sure that QUERY connection will timeout first
+            def get_connection_holders(self):
+                holders = super(PatchedCluster, self).get_connection_holders()
+                return sorted(holders, reverse=True, key=lambda v: int(v._connection.host == query_host))
+
+            # Block executor thread like closing a dead socket could do
+            def connection_factory(self, *args, **kwargs):
+                conn = super(PatchedCluster, self).connection_factory(*args, **kwargs)
+                conn.defunct = late(seconds=2*lag)(conn.defunct)
+                return conn
+
+        cluster = PatchedCluster(
+            protocol_version=PROTOCOL_VERSION,
+            compression=False,
+            idle_heartbeat_interval=heartbeat_interval,
+            idle_heartbeat_timeout=heartbeat_timeout,
+            load_balancing_policy=PatchedRoundRobinPolicy()
+        )
+        session = cluster.connect()
+        self.addCleanup(cluster.shutdown)
+
+        prime_query(query_to_prime, then={"delay_in_ms": never})
+
+        # Make heartbeat due
+        time.sleep(heartbeat_interval)
+
+        future = session.execute_async(query_to_prime, timeout=heartbeat_interval+heartbeat_timeout+3*lag)
+        # Delay thread execution like kernel could do
+        future._retry_task = late(seconds=4*lag)(future._retry_task)
+
+        prime_request(PrimeOptions(then={"result": "no_result", "delay_in_ms": never}))
+        prime_request(RejectConnections("unbind"))
+
+        self.assertRaisesRegexp(OperationTimedOut, "Connection defunct by heartbeat", future.result)
+
     def test_close_when_query(self):
         """
         Test to ensure the driver behaves correctly if the connection is closed

tests/integration/upgrade/test_upgrade.py

@@ -27,7 +27,7 @@
 
 
 two_to_three_path = upgrade_paths([
-    UpgradePath("2.2.9-3.11", {"version": "2.2.9"}, {"version": "3.11"}, {}),
+    UpgradePath("2.2.9-3.11", {"version": "2.2.9"}, {"version": "3.11.4"}, {}),
 ])
 class UpgradeTests(UpgradeBase):
     @two_to_three_path
@@ -176,7 +176,7 @@ def _assert_same_token_map(self, original, new):
 
 
 two_to_three_with_auth_path = upgrade_paths([
-    UpgradePath("2.2.9-3.11-auth", {"version": "2.2.9"}, {"version": "3.11"},
+    UpgradePath("2.2.9-3.11-auth", {"version": "2.2.9"}, {"version": "3.11.4"},
                 {'authenticator': 'PasswordAuthenticator',
                  'authorizer': 'CassandraAuthorizer'}),
 ])

tests/integration/util.py

@@ -13,6 +13,7 @@
 # limitations under the License.
 
 from tests.integration import PROTOCOL_VERSION
+from functools import wraps
 import time
 
 
@@ -96,3 +97,13 @@ def wrapped_condition():
 
     # last attempt, let the exception raise
     condition()
+
+
+def late(seconds=1):
+    def decorator(func):
+        @wraps(func)
+        def wrapper(*args, **kwargs):
+            time.sleep(seconds)
+            func(*args, **kwargs)
+        return wrapper
+    return decorator

tests/unit/test_response_future.py

@@ -19,7 +19,7 @@
 
 from mock import Mock, MagicMock, ANY
 
-from cassandra import ConsistencyLevel, Unavailable, SchemaTargetType, SchemaChangeType
+from cassandra import ConsistencyLevel, Unavailable, SchemaTargetType, SchemaChangeType, OperationTimedOut
 from cassandra.cluster import Session, ResponseFuture, NoHostAvailable, ProtocolVersion
 from cassandra.connection import Connection, ConnectionException
 from cassandra.protocol import (ReadTimeoutErrorMessage, WriteTimeoutErrorMessage,
@@ -125,6 +125,39 @@ def test_other_result_message_kind(self):
         rf._set_result(None, None, None, Mock(spec=ResultMessage, kind=999, results=result))
         self.assertListEqual(list(rf.result()), result)
 
+    def test_heartbeat_defunct_deadlock(self):
+        """
+        Heartbeat defuncts all connections and clears request queues. Response future times out and even
+        if it has been removed from request queue, timeout exception must be thrown. Otherwise event loop
+        will deadlock on eventual ResponseFuture.result() call.
+
+        PYTHON-1044
+        """
+
+        connection = MagicMock(spec=Connection)
+        connection._requests = {}
+
+        pool = Mock()
+        pool.is_shutdown = False
+        pool.borrow_connection.return_value = [connection, 1]
+
+        session = self.make_basic_session()
+        session.cluster._default_load_balancing_policy.make_query_plan.return_value = [Mock(), Mock()]
+        session._pools.get.return_value = pool
+
+        query = SimpleStatement("SELECT * FROM foo")
+        message = QueryMessage(query=query, consistency_level=ConsistencyLevel.ONE)
+
+        rf = ResponseFuture(session, message, query, 1)
+        rf.send_request()
+
+        # Simulate Connection.error_all_requests() after heartbeat defuncts
+        connection._requests = {}
+
+        # Simulate ResponseFuture timing out
+        rf._on_timeout()
+        self.assertRaisesRegexp(OperationTimedOut, "Connection defunct by heartbeat", rf.result)
+
     def test_read_timeout_error_message(self):
         session = self.make_session()
         query = SimpleStatement("SELECT * FROM foo")