Don't call join() inside close()

Author: NoahStapp

pymongo/asynchronous/mongo_client.py

@@ -1559,13 +1559,13 @@ async def close(self) -> None:
         # Stop the periodic task thread and then send pending killCursor
         # requests before closing the topology.
         self._kill_cursors_executor.close()
+        if not _IS_SYNC:
+            await self._kill_cursors_executor.join()
         await self._process_kill_cursors()
         await self._topology.close()
         if self._encrypter:
             # TODO: PYTHON-1921 Encrypted MongoClients cannot be re-opened.
             await self._encrypter.close()
-        if not _IS_SYNC:
-            await self._kill_cursors_executor.join()
         self._closed = True
 
     if not _IS_SYNC:

pymongo/asynchronous/monitor.py

@@ -111,8 +111,6 @@ async def close(self) -> None:
         open() restarts the monitor after closing.
         """
         self.gc_safe_close()
-        if not _IS_SYNC:
-            await self._executor.join()
 
     async def join(self, timeout: Optional[int] = None) -> None:
         """Wait for the monitor to stop."""
@@ -191,10 +189,12 @@ def gc_safe_close(self) -> None:
         self._rtt_monitor.gc_safe_close()
         self.cancel_check()
 
+    async def join(self, timeout: Optional[int] = None) -> None:
+        await self._executor.join(timeout)
+        await self._rtt_monitor.join()
+
     async def close(self) -> None:
         self.gc_safe_close()
-        if not _IS_SYNC:
-            await self._executor.join()
         await self._rtt_monitor.close()
         # Increment the generation and maybe close the socket. If the executor
         # thread has the socket checked out, it will be closed when checked in.
@@ -464,8 +464,6 @@ async def close(self) -> None:
         self.gc_safe_close()
         # Increment the generation and maybe close the socket. If the executor
         # thread has the socket checked out, it will be closed when checked in.
-        if not _IS_SYNC:
-            await self._executor.join()
         await self._pool.reset()
 
     async def add_sample(self, sample: float) -> None:

pymongo/asynchronous/topology.py

@@ -520,6 +520,8 @@ async def _process_change(
             and self._description.topology_type not in SRV_POLLING_TOPOLOGIES
         ):
             await self._srv_monitor.close()
+            if not _IS_SYNC:
+                await self._srv_monitor.join()
 
         # Clear the pool from a failed heartbeat.
         if reset_pool:

pymongo/synchronous/mongo_client.py

@@ -1553,13 +1553,13 @@ def close(self) -> None:
         # Stop the periodic task thread and then send pending killCursor
         # requests before closing the topology.
         self._kill_cursors_executor.close()
+        if not _IS_SYNC:
+            self._kill_cursors_executor.join()
         self._process_kill_cursors()
         self._topology.close()
         if self._encrypter:
             # TODO: PYTHON-1921 Encrypted MongoClients cannot be re-opened.
             self._encrypter.close()
-        if not _IS_SYNC:
-            self._kill_cursors_executor.join()
         self._closed = True
 
     if not _IS_SYNC:

pymongo/synchronous/monitor.py

@@ -111,8 +111,6 @@ def close(self) -> None:
         open() restarts the monitor after closing.
         """
         self.gc_safe_close()
-        if not _IS_SYNC:
-            self._executor.join()
 
     def join(self, timeout: Optional[int] = None) -> None:
         """Wait for the monitor to stop."""
@@ -191,10 +189,12 @@ def gc_safe_close(self) -> None:
         self._rtt_monitor.gc_safe_close()
         self.cancel_check()
 
+    def join(self, timeout: Optional[int] = None) -> None:
+        self._executor.join(timeout)
+        self._rtt_monitor.join()
+
     def close(self) -> None:
         self.gc_safe_close()
-        if not _IS_SYNC:
-            self._executor.join()
         self._rtt_monitor.close()
         # Increment the generation and maybe close the socket. If the executor
         # thread has the socket checked out, it will be closed when checked in.
@@ -464,8 +464,6 @@ def close(self) -> None:
         self.gc_safe_close()
         # Increment the generation and maybe close the socket. If the executor
         # thread has the socket checked out, it will be closed when checked in.
-        if not _IS_SYNC:
-            self._executor.join()
         self._pool.reset()
 
     def add_sample(self, sample: float) -> None:

pymongo/synchronous/topology.py

@@ -520,6 +520,8 @@ def _process_change(
             and self._description.topology_type not in SRV_POLLING_TOPOLOGIES
         ):
             self._srv_monitor.close()
+            if not _IS_SYNC:
+                self._srv_monitor.join()
 
         # Clear the pool from a failed heartbeat.
         if reset_pool: