more fixes to asyncio

Author: fruch

cassandra/connection.py

@@ -940,7 +940,7 @@ def _connect_socket(self):
                 self._initiate_connection(sockaddr)
                 self._socket.settimeout(None)
                 local_addr = self._socket.getsockname()
-                log.debug('Connection %s %s:%s -> %s:%s', id(self), local_addr[0], local_addr[1], sockaddr[0], sockaddr[1])
+                # log.debug('Connection %s %s:%s -> %s:%s', id(self), local_addr[0], local_addr[1], sockaddr[0], sockaddr[1])
                 if self._check_hostname:
                     self._match_hostname()
                 sockerr = None

cassandra/io/asyncioreactor.py

@@ -1,3 +1,5 @@
+import threading
+
 from cassandra.connection import Connection, ConnectionShutdown
 import sys
 import asyncio
@@ -41,13 +43,12 @@ def end(self):
 
     def __init__(self, timeout, callback, loop):
         delayed = self._call_delayed_coro(timeout=timeout,
-                                          callback=callback,
-                                          loop=loop)
+                                          callback=callback)
         self._handle = asyncio.run_coroutine_threadsafe(delayed, loop=loop)
 
     @staticmethod
-    async def _call_delayed_coro(timeout, callback, loop):
-        await asyncio.sleep(timeout, loop=loop)
+    async def _call_delayed_coro(timeout, callback):
+        await asyncio.sleep(timeout)
         return callback()
 
     def __lt__(self, other):
@@ -111,8 +112,11 @@ def initialize_reactor(cls):
             if cls._pid != os.getpid():
                 cls._loop = None
             if cls._loop is None:
-                cls._loop = asyncio.new_event_loop()
-                asyncio.set_event_loop(cls._loop)
+                try:
+                    cls._loop = asyncio.get_running_loop()
+                except RuntimeError:
+                    cls._loop = asyncio.new_event_loop()
+                    asyncio.set_event_loop(cls._loop)
 
             if not cls._loop_thread:
                 # daemonize so the loop will be shut down on interpreter
@@ -165,7 +169,7 @@ def push(self, data):
         else:
             chunks = [data]
 
-        if self._loop_thread.ident != get_ident():
+        if self._loop_thread != threading.current_thread():
             asyncio.run_coroutine_threadsafe(
                 self._push_msg(chunks),
                 loop=self._loop