Merge pull request apache#899 from datastax/python-853_retry-implementation

mambocab · web-flow · commit b7d299480621 · 2018-01-23T16:06:57.000-05:00
PYTHON-853: delay timeout
diff --git a/CHANGELOG.rst b/CHANGELOG.rst
@@ -12,6 +12,7 @@ Bug Fixes
 * Support retry_policy in PreparedStatement (PYTHON-861)
 * __del__ method in Session is throwing an exception (PYTHON-813)
 * LZ4 import issue with recent versions (PYTHON-897)
+* ResponseFuture._connection can be None when returning request_id (PYTHON-853)
 
 Other
 -----
diff --git a/cassandra/cluster.py b/cassandra/cluster.py
@@ -3501,22 +3501,37 @@ def _cancel_timer(self):
         if self._timer:
             self._timer.cancel()
 
-    def _on_timeout(self):
+    def _on_timeout(self, _attempts=0):
+        """
+        Called when the request associated with this ResponseFuture times out.
 
-        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
-        except KeyError:
+        This function may reschedule itself. The ``_attempts`` parameter tracks
+        the number of times this has happened. This parameter should only be
+        set in those cases, where ``_on_timeout`` reschedules itself.
+        """
+        # PYTHON-853: for short timeouts, we sometimes race with our __init__
+        if self._connection is None and _attempts < 3:
+            self._timer = self.session.cluster.connection_class.create_timer(
+                0.01,
+                partial(self._on_timeout, _attempts=_attempts + 1)
+            )
             return
 
-        pool = self.session._pools.get(self._current_host)
-        if pool and not pool.is_shutdown:
-            with self._connection.lock:
-                self._connection.request_ids.append(self._req_id)
+        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
+            except KeyError:
+                return
+
+            pool = self.session._pools.get(self._current_host)
+            if pool and not pool.is_shutdown:
+                with self._connection.lock:
+                    self._connection.request_ids.append(self._req_id)
 
-            pool.return_connection(self._connection)
+                pool.return_connection(self._connection)
 
         errors = self._errors
         if not errors:
