PYTHON-1898 Close events executor faster after the topology is freed

Author: ShaneHarvey

pymongo/topology.py

@@ -74,7 +74,7 @@ def __init__(self, topology_settings):
 
         # Create events queue if there are publishers.
         self._events = None
-        self._events_thread = None
+        self.__events_executor = None
 
         if self._publish_server or self._publish_tp:
             self._events = Queue.Queue(maxsize=100)
@@ -127,7 +127,7 @@ def target():
             # We strongly reference the executor and it weakly references
             # the queue via this closure. When the topology is freed, stop
             # the executor soon.
-            weak = weakref.ref(self._events)
+            weak = weakref.ref(self._events, executor.close)
             self.__events_executor = executor
             executor.open()
 

test/test_monitor.py

@@ -21,29 +21,64 @@
 sys.path[0:0] = [""]
 
 from pymongo.periodic_executor import _EXECUTORS
-from test import client_context, unittest, IntegrationTest
-from test.utils import single_client, one, connected, wait_until
+
+from test import unittest, IntegrationTest
+from test.utils import (connected,
+                        ServerAndTopologyEventListener,
+                        single_client,
+                        wait_until)
 
 
 def unregistered(ref):
     gc.collect()
     return ref not in _EXECUTORS
 
 
+def get_executors(client):
+    executors = []
+    for server in client._topology._servers.values():
+        executors.append(server._monitor._executor)
+    executors.append(client._kill_cursors_executor)
+    executors.append(client._topology._Topology__events_executor)
+    return [e for e in executors if e is not None]
+
+
+def create_client():
+    listener = ServerAndTopologyEventListener()
+    client = single_client(event_listeners=[listener])
+    connected(client)
+    return client
+
+
 class TestMonitor(IntegrationTest):
-    def test_atexit_hook(self):
-        client = single_client(client_context.host, client_context.port)
-        executor = one(client._topology._servers.values())._monitor._executor
-        connected(client)
+    def test_cleanup_executors_on_client_del(self):
+        client = create_client()
+        executors = get_executors(client)
+        self.assertEqual(len(executors), 3)
 
-        # The executor stores a weakref to itself in _EXECUTORS.
-        ref = one([r for r in _EXECUTORS.copy() if r() is executor])
+        # Each executor stores a weakref to itself in _EXECUTORS.
+        executor_refs = [
+            (r, r()._name) for r in _EXECUTORS.copy() if r() in executors]
 
-        del executor
+        del executors
         del client
 
-        wait_until(partial(unregistered, ref), 'unregister executor',
-                   timeout=5)
+        for ref, name in executor_refs:
+            wait_until(partial(unregistered, ref),
+                       'unregister executor: %s' % (name,),
+                       timeout=5)
+
+    def test_cleanup_executors_on_client_close(self):
+        client = create_client()
+        executors = get_executors(client)
+        self.assertEqual(len(executors), 3)
+
+        client.close()
+
+        for executor in executors:
+            wait_until(lambda: executor._stopped,
+                       'closed executor: %s' % (executor._name,),
+                       timeout=5)
 
 
 if __name__ == "__main__":