Merge pull request apache#891 from datastax/python-836-2

beltran · web-flow · commit 8ac3fbb26dfa · 2018-01-10T13:06:13.000-05:00
Alternative solution for PYTHON-836
diff --git a/CHANGELOG.rst b/CHANGELOG.rst
@@ -36,6 +36,7 @@ Features
 * Include hash of result set metadata in prepared stmt id (PYTHON-808)
 * Add NO_COMPACT startup option (PYTHON-839)
 * Add new exception type for CDC (PYTHON-837)
+* Allow 0ms in ConstantSpeculativeExecutionPolicy (PYTHON-836)
 
 Bug Fixes
 ---------
diff --git a/cassandra/cluster.py b/cassandra/cluster.py
@@ -3527,6 +3527,18 @@ def _on_timeout(self):
     def _on_speculative_execute(self):
         self._timer = None
         if not self._event.is_set():
+
+            # PYTHON-836, the speculative queries must be after
+            # the query is sent from the main thread, otherwise the
+            # query from the main thread may raise NoHostAvailable
+            # if the _query_plan has been exhausted by the specualtive queries.
+            # This also prevents a race condition accessing the iterator.
+            # We reschedule this call until the main thread has succeeded
+            # making a query
+            if not self.attempted_hosts:
+                self._timer = self.session.cluster.connection_class.create_timer(0.01, self._on_speculative_execute)
+                return
+
             if self._time_remaining is not None:
                 if self._time_remaining <= 0:
                     self._on_timeout()
@@ -3540,10 +3552,6 @@ def _make_query_plan(self):
         # calls to send_request (which retries may do) will resume where
         # they last left off
         self.query_plan = iter(self._load_balancer.make_query_plan(self.session.keyspace, self.query))
-        # Make iterator thread safe when there can be a speculative delay since it could
-        # from different threads
-        if isinstance(self._spec_execution_plan, NoSpeculativeExecutionPlan):
-            self.query_plan = _threadsafe_iter(self.query_plan)
 
     def send_request(self, error_no_hosts=True):
         """ Internal """
@@ -4316,16 +4324,3 @@ def paging_state(self):
         avoid sending this to untrusted parties.
         """
         return self.response_future._paging_state
-
-
-class _threadsafe_iter(six.Iterator):
-    def __init__(self, it):
-        self.it = it
-        self.lock = Lock()
-
-    def __iter__(self):
-        return self
-
-    def __next__(self):
-        with self.lock:
-            return next(self.it)
diff --git a/tests/integration/simulacron/test_policies.py b/tests/integration/simulacron/test_policies.py
@@ -17,7 +17,7 @@
     import unittest  # noqa
 
 from cassandra import OperationTimedOut, WriteTimeout
-from cassandra.cluster import Cluster, ExecutionProfile
+from cassandra.cluster import Cluster, ExecutionProfile, ResponseFuture
 from cassandra.query import SimpleStatement
 from cassandra.policies import ConstantSpeculativeExecutionPolicy, RoundRobinPolicy, RetryPolicy, WriteType
 
@@ -168,6 +168,45 @@ def test_speculative_and_timeout(self):
         # This is because 14 / 4 + 1 = 4
         self.assertEqual(len(response_future.attempted_hosts), 4)
 
+    def test_delay_can_be_0(self):
+        """
+        Test to validate that the delay can be zero for the ConstantSpeculativeExecutionPolicy
+        @since 3.13
+        @jira_ticket PYTHON-836
+        @expected_result all the queries are executed immediately
+        @test_category policy
+        """
+        query_to_prime = "INSERT INTO madeup_keyspace.madeup_table(k, v) VALUES (1, 2)"
+        prime_query(query_to_prime, then={"delay_in_ms": 5000})
+        number_of_requests = 4
+        spec = ExecutionProfile(load_balancing_policy=RoundRobinPolicy(),
+                                speculative_execution_policy=ConstantSpeculativeExecutionPolicy(0, number_of_requests))
+
+        cluster = Cluster()
+        cluster.add_execution_profile("spec", spec)
+        session = cluster.connect(wait_for_all_pools=True)
+        self.addCleanup(cluster.shutdown)
+
+        counter = count()
+
+        def patch_and_count(f):
+            def patched(*args, **kwargs):
+                next(counter)
+                print("patched")
+                f(*args, **kwargs)
+            return patched
+
+        self.addCleanup(setattr, ResponseFuture, "send_request", ResponseFuture.send_request)
+        ResponseFuture.send_request = patch_and_count(ResponseFuture.send_request)
+        stmt = SimpleStatement(query_to_prime)
+        stmt.is_idempotent = True
+        results = session.execute(stmt, execution_profile="spec")
+        self.assertEqual(len(results.response_future.attempted_hosts), 3)
+
+        # send_request is called number_of_requests times for the speculative request
+        # plus one for the call from the main thread.
+        self.assertEqual(next(counter), number_of_requests + 1)
+
 
 class CustomRetryPolicy(RetryPolicy):
     def on_write_timeout(self, query, consistency, write_type,
@@ -193,7 +232,6 @@ def on_read_timeout(self, query, consistency, required_responses,
 
     def on_write_timeout(self, query, consistency, write_type,
                          required_responses, received_responses, retry_num):
-        print("counter on_write_timeout")
         next(self.write_timeout)
         return self.IGNORE, None
 
diff --git a/tests/integration/standard/test_policies.py b/tests/integration/standard/test_policies.py
@@ -19,14 +19,11 @@
 
 from cassandra.cluster import Cluster, ExecutionProfile, ResponseFuture
 from cassandra.policies import HostFilterPolicy, RoundRobinPolicy,  SimpleConvictionPolicy, \
-    WhiteListRoundRobinPolicy, ConstantSpeculativeExecutionPolicy
+    WhiteListRoundRobinPolicy
 from cassandra.pool import Host
-from cassandra.query import SimpleStatement
 
 from tests.integration import PROTOCOL_VERSION, local, use_singledc
-from tests import notwindows
 
-from itertools import count
 from concurrent.futures import wait as wait_futures
 
 def setup_module():
@@ -92,37 +89,3 @@ def test_only_connects_to_subset(self):
             queried_hosts.update(response.response_future.attempted_hosts)
         queried_hosts = set(host.address for host in queried_hosts)
         self.assertEqual(queried_hosts, only_connect_hosts)
-
-
-class SpeculativeExecutionPolicy(unittest.TestCase):
-    @notwindows
-    def test_delay_can_be_0(self):
-        """
-        Test to validate that the delay can be zero for the ConstantSpeculativeExecutionPolicy
-        @since 3.13
-        @jira_ticket PYTHON-836
-        @expected_result all the queries are executed immediately
-        @test_category policy
-        """
-        number_of_requests = 4
-        spec = ExecutionProfile(speculative_execution_policy=ConstantSpeculativeExecutionPolicy(0, number_of_requests))
-
-        cluster = Cluster()
-        cluster.add_execution_profile("spec", spec)
-        session = cluster.connect(wait_for_all_pools=True)
-        self.addCleanup(cluster.shutdown)
-
-        counter = count()
-
-        def patch_and_count(f):
-            def patched(*args, **kwargs):
-                next(counter)
-                f(*args, **kwargs)
-            return patched
-
-        ResponseFuture._on_speculative_execute = patch_and_count(ResponseFuture._on_speculative_execute)
-        stmt = SimpleStatement("INSERT INTO test3rf.test(k, v) VALUES (1, 2)")
-        stmt.is_idempotent = True
-        results = session.execute(stmt, execution_profile="spec")
-        self.assertEqual(len(results.response_future.attempted_hosts), 3)
-        self.assertEqual(next(counter), number_of_requests)
