PYTHON-1704 Close periodic task thread when client is closed (#389)

pymongo/mongo_client.py

@@ -1030,8 +1030,9 @@ class MongoClient(common.BaseObject):
         session_ids = self._topology.pop_all_sessions()
         if session_ids:
             self._end_sessions(session_ids)
-        # Run _process_periodic_tasks to send pending killCursor requests
-        # before closing the topology.
+        # Stop the periodic task thread and then run _process_periodic_tasks
+        # to send pending killCursor requests before closing the topology.
+        self._kill_cursors_executor.close()
         self._process_periodic_tasks()
         self._topology.close()
 
@@ -1071,6 +1072,8 @@ class MongoClient(common.BaseObject):
         launches the connection process in the background.
         """
         self._topology.open()
+        with self.__lock:
+            self._kill_cursors_executor.open()
         return self._topology
 
     @contextlib.contextmanager
@@ -1136,10 +1139,6 @@ class MongoClient(common.BaseObject):
           - `address` (optional): Optional address when sending a message
             to a specific server, used for getMore.
         """
-        with self.__lock:
-            # If needed, restart kill-cursors thread after a fork.
-            self._kill_cursors_executor.open()
-
         topology = self._get_topology()
         if address:
             server = topology.select_server_by_address(address)

test/test_client.py

@@ -640,6 +640,23 @@ class TestClient(IntegrationTest):
         self.client._process_periodic_tasks()
         self.assertFalse(self.client._topology._opened)
 
+    def test_close_stops_kill_cursors_thread(self):
+        client = rs_client()
+        client.test.test.find_one()
+        self.assertFalse(client._kill_cursors_executor._stopped)
+
+        # Closing the client should stop the thread.
+        client.close()
+        self.assertTrue(client._kill_cursors_executor._stopped)
+
+        # Reusing the closed client should restart the thread.
+        client.admin.command('isMaster')
+        self.assertFalse(client._kill_cursors_executor._stopped)
+
+        # Again, closing the client should stop the thread.
+        client.close()
+        self.assertTrue(client._kill_cursors_executor._stopped)
+
     def test_bad_uri(self):
         with self.assertRaises(InvalidURI):
             MongoClient("http://localhost")