Move calls to asyncore to the loop thread

bjmb · bjmb · commit 10ecc3721242 · 2017-07-31T17:59:36.000-04:00
diff --git a/build.yaml b/build.yaml
@@ -14,11 +14,11 @@ schedules:
     branches:
       include: [/python.*/]
     env_vars: |
-      EVENT_LOOP_MANAGER='libev'
+      EVENT_LOOP_MANAGER='async'
     matrix:
       exclude:
         - python: [3.4, 3.6]
-        - cassandra: ['2.0', '2.1', '3.0']
+        - cassandra: ['2.0', '2.1']
 
   weekly_libev:
     schedule: 0 10 * * 6
diff --git a/cassandra/io/asyncorereactor.py b/cassandra/io/asyncorereactor.py
@@ -18,9 +18,10 @@
 import os
 import socket
 import sys
-from threading import Lock, Thread
+from threading import Lock, Thread, Event
 import time
 import weakref
+import sys
 
 from six.moves import range
 
@@ -51,6 +52,33 @@ def _cleanup(loop_weakref):
     loop._cleanup()
 
 
+class WaitableTimer(Timer):
+    def __init__(self, timeout, callback):
+        Timer.__init__(self, timeout, callback)
+        self.callback = callback
+        self.event = Event()
+
+        self.final_exception = None
+
+    def finish(self, time_now):
+        try:
+            finished = Timer.finish(self, time_now)
+            if finished:
+                self.event.set()
+                return True
+            return False
+
+        except Exception as e:
+            self.final_exception = e
+            self.event.set()
+            return True
+
+    def wait(self, timeout=None):
+        self.event.wait(timeout)
+        if self.final_exception:
+            raise self.final_exception
+
+
 class _PipeWrapper(object):
 
     def __init__(self, fd):
@@ -239,6 +267,11 @@ def _run_loop(self):
     def add_timer(self, timer):
         self._timers.add_timer(timer)
 
+        # This function is called from a different thread than the event loop
+        # thread, so for this call to be thread safe, we must wake up the loop
+        # in case it's stuck at a select
+        self.wake_loop()
+
     def _cleanup(self):
         global _dispatcher_map
 
@@ -305,16 +338,23 @@ def __init__(self, *args, **kwargs):
         self.deque_lock = Lock()
 
         self._connect_socket()
-        asyncore.dispatcher.__init__(self, self._socket, _dispatcher_map)
+
+        # start the event loop if needed
+        self._loop.maybe_start()
+
+        init_handler = WaitableTimer(
+            timeout=0,
+            callback=partial(asyncore.dispatcher.__init__,
+                             self, self._socket, _dispatcher_map)
+        )
+        AsyncoreConnection._loop.add_timer(init_handler)
+        init_handler.wait(kwargs["connect_timeout"])
 
         self._writable = True
         self._readable = True
 
         self._send_options_message()
 
-        # start the event loop if needed
-        self._loop.maybe_start()
-
     def close(self):
         with self.lock:
             if self.is_closed:
@@ -324,7 +364,10 @@ def close(self):
         log.debug("Closing connection (%s) to %s", id(self), self.host)
         self._writable = False
         self._readable = False
-        asyncore.dispatcher.close(self)
+
+        # We don't have to wait for this to be closed, we can just schedule it
+        AsyncoreConnection.create_timer(0, partial(asyncore.dispatcher.close, self))
+
         log.debug("Closed socket to %s", self.host)
 
         if not self.is_defunct:
diff --git a/tests/integration/__init__.py b/tests/integration/__init__.py
@@ -265,6 +265,8 @@ def get_unsupported_upper_protocol():
 dseonly = unittest.skipUnless(DSE_VERSION, "Test is only applicalbe to DSE clusters")
 pypy = unittest.skipUnless(platform.python_implementation() == "PyPy", "Test is skipped unless it's on PyPy")
 notpy3 = unittest.skipIf(sys.version_info >= (3, 0), "Test not applicable for Python 3.x runtime")
+requiresmallclockgranularity = unittest.skipIf("Windows" in platform.system() or "async" in EVENT_LOOP_MANAGER,
+                                               "This test is not suitible for environments with large clock granularity")
 
 
 def wait_for_node_socket(node, timeout):
diff --git a/tests/integration/long/test_failure_types.py b/tests/integration/long/test_failure_types.py
@@ -20,7 +20,8 @@
 from cassandra.policies import HostFilterPolicy, RoundRobinPolicy
 from cassandra.concurrent import execute_concurrent_with_args
 from cassandra.query import SimpleStatement
-from tests.integration import use_singledc, PROTOCOL_VERSION, get_cluster, setup_keyspace, remove_cluster, get_node
+from tests.integration import use_singledc, PROTOCOL_VERSION, get_cluster, setup_keyspace, remove_cluster, get_node, \
+    requiresmallclockgranularity
 from mock import Mock
 
 try:
@@ -318,6 +319,7 @@ def test_user_function_failure(self):
             """, consistency_level=ConsistencyLevel.ALL, expected_exception=None)
 
 
+@requiresmallclockgranularity
 class TimeoutTimerTest(unittest.TestCase):
     def setUp(self):
         """
diff --git a/tests/integration/standard/test_connection.py b/tests/integration/standard/test_connection.py
@@ -32,9 +32,8 @@
 from cassandra.policies import HostFilterPolicy, RoundRobinPolicy, HostStateListener
 from cassandra.pool import HostConnectionPool
 
-from tests import is_monkey_patched, notwindows
-from tests.integration import use_singledc, PROTOCOL_VERSION, get_node, CASSANDRA_IP, local
-
+from tests import is_monkey_patched
+from tests.integration import use_singledc, PROTOCOL_VERSION, get_node, CASSANDRA_IP, local, requiresmallclockgranularity
 try:
     from cassandra.io.libevreactor import LibevConnection
 except ImportError:
@@ -364,9 +363,7 @@ def send_msgs(conn, event):
         for t in threads:
             t.join()
 
-    # We skip this one for windows because the clock is not as
-    # granular as in linux
-    @notwindows
+    @requiresmallclockgranularity
     def test_connect_timeout(self):
         # Underlying socket implementations don't always throw a socket timeout even with min float
         # This can be timing sensitive, added retry to ensure failure occurs if it can
diff --git a/tests/integration/standard/test_policies.py b/tests/integration/standard/test_policies.py
@@ -28,7 +28,7 @@
 from cassandra.pool import Host
 
 from tests.integration import BasicSharedKeyspaceUnitTestCase, greaterthancass21, PROTOCOL_VERSION
-from tests import notwindows
+from tests.integration import requiresmallclockgranularity
 
 from mock import patch
 
@@ -92,8 +92,7 @@ def test_predicate_changes(self):
         self.assertEqual(queried_hosts, all_hosts)
 
 
-# This doesn't work well with Windows clock granularity
-@notwindows
+@requiresmallclockgranularity
 class SpecExecTest(BasicSharedKeyspaceUnitTestCase):
 
     @classmethod
diff --git a/tests/unit/io/test_asyncorereactor.py b/tests/unit/io/test_asyncorereactor.py
@@ -63,7 +63,7 @@ def setUp(self):
             raise unittest.SkipTest("Can't test asyncore with monkey patching")
 
     def make_connection(self):
-        c = AsyncoreConnection('1.2.3.4', cql_version='3.0.1')
+        c = AsyncoreConnection('1.2.3.4', cql_version='3.0.1', connect_timeout=5)
         c.socket = Mock()
         c.socket.send.side_effect = lambda x: len(x)
         return c
