PYTHON-973: Use global variable for libev loops so it can be subclassed

Author: beltran

CHANGELOG.rst

@@ -4,6 +4,10 @@ Bug Fixes
 ---------
 * Tokenmap.get_replicas returns the wrong value if token coincides with the end of the range (PYTHON-978)
 
+Other
+-----
+* Use global variable for libev loops so it can be subclassed (PYTHON-973)
+
 3.14.0
 ======
 April 17, 2018

cassandra/io/libevreactor.py

@@ -41,12 +41,9 @@
 log = logging.getLogger(__name__)
 
 
-def _cleanup(loop_weakref):
-    try:
-        loop = loop_weakref()
-    except ReferenceError:
-        return
-    loop._cleanup()
+def _cleanup(loop):
+    if loop:
+        loop._cleanup()
 
 
 class LibevLoop(object):
@@ -84,8 +81,6 @@ def __init__(self):
         self._timers = TimerManager()
         self._loop_timer = libev.Timer(self._loop, self._on_loop_timer)
 
-        atexit.register(partial(_cleanup, weakref.ref(self)))
-
     def maybe_start(self):
         should_start = False
         with self._lock:
@@ -228,36 +223,41 @@ def _loop_will_run(self, prepare):
             self._notifier.send()
 
 
+_global_loop = None
+atexit.register(partial(_cleanup, _global_loop))
+
+
 class LibevConnection(Connection):
     """
     An implementation of :class:`.Connection` that uses libev for its event loop.
     """
-    _libevloop = None
     _write_watcher_is_active = False
     _read_watcher = None
     _write_watcher = None
     _socket = None
 
     @classmethod
     def initialize_reactor(cls):
-        if not cls._libevloop:
-            cls._libevloop = LibevLoop()
+        global _global_loop
+        if not _global_loop:
+            _global_loop = LibevLoop()
         else:
-            if cls._libevloop._pid != os.getpid():
+            if _global_loop._pid != os.getpid():
                 log.debug("Detected fork, clearing and reinitializing reactor state")
                 cls.handle_fork()
-                cls._libevloop = LibevLoop()
+                _global_loop = LibevLoop()
 
     @classmethod
     def handle_fork(cls):
-        if cls._libevloop:
-            cls._libevloop._cleanup()
-            cls._libevloop = None
+        global _global_loop
+        if _global_loop:
+            _global_loop._cleanup()
+            _global_loop = None
 
     @classmethod
     def create_timer(cls, timeout, callback):
         timer = Timer(timeout, callback)
-        cls._libevloop.add_timer(timer)
+        _global_loop.add_timer(timer)
         return timer
 
     def __init__(self, *args, **kwargs):
@@ -268,16 +268,16 @@ def __init__(self, *args, **kwargs):
         self._connect_socket()
         self._socket.setblocking(0)
 
-        with self._libevloop._lock:
-            self._read_watcher = libev.IO(self._socket.fileno(), libev.EV_READ, self._libevloop._loop, self.handle_read)
-            self._write_watcher = libev.IO(self._socket.fileno(), libev.EV_WRITE, self._libevloop._loop, self.handle_write)
+        with _global_loop._lock:
+            self._read_watcher = libev.IO(self._socket.fileno(), libev.EV_READ, _global_loop._loop, self.handle_read)
+            self._write_watcher = libev.IO(self._socket.fileno(), libev.EV_WRITE, _global_loop._loop, self.handle_write)
 
         self._send_options_message()
 
-        self._libevloop.connection_created(self)
+        _global_loop.connection_created(self)
 
         # start the global event loop if needed
-        self._libevloop.maybe_start()
+        _global_loop.maybe_start()
 
     def close(self):
         with self.lock:
@@ -286,7 +286,8 @@ def close(self):
             self.is_closed = True
 
         log.debug("Closing connection (%s) to %s", id(self), self.host)
-        self._libevloop.connection_destroyed(self)
+
+        _global_loop.connection_destroyed(self)
         self._socket.close()
         log.debug("Closed socket to %s", self.host)
 
@@ -367,4 +368,4 @@ def push(self, data):
 
         with self._deque_lock:
             self.deque.extend(chunks)
-            self._libevloop.notify()
+            _global_loop.notify()