PYTHON-5053 - AsyncMongoClient.close() should await all background tasks

Author: NoahStapp

pymongo/asynchronous/mongo_client.py

@@ -1564,6 +1564,8 @@ async def close(self) -> None:
         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,6 +111,8 @@ 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,6 +193,8 @@ def gc_safe_close(self) -> None:
 
     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.
@@ -460,6 +464,8 @@ 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/server.py

@@ -115,6 +115,8 @@ async def close(self) -> None:
             )
 
         await self._monitor.close()
+        if not _IS_SYNC:
+            await self._monitor.join()
         await self._pool.close()
 
     def request_check(self) -> None:

pymongo/asynchronous/topology.py

@@ -705,6 +705,8 @@ async def close(self) -> None:
             # Stop SRV polling thread.
             if self._srv_monitor:
                 await self._srv_monitor.close()
+                if not _IS_SYNC:
+                    await self._srv_monitor.join()
 
             self._opened = False
             self._closed = True

pymongo/periodic_executor.py

@@ -75,6 +75,8 @@ def close(self, dummy: Any = None) -> None:
         callback; see monitor.py.
         """
         self._stopped = True
+        if self._task is not None:
+            self._task.cancel()
 
     async def join(self, timeout: Optional[int] = None) -> None:
         if self._task is not None:

pymongo/synchronous/mongo_client.py

@@ -1558,6 +1558,8 @@ def close(self) -> None:
         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,6 +111,8 @@ 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,6 +193,8 @@ def gc_safe_close(self) -> None:
 
     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.
@@ -460,6 +464,8 @@ 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/server.py

@@ -115,6 +115,8 @@ def close(self) -> None:
             )
 
         self._monitor.close()
+        if not _IS_SYNC:
+            self._monitor.join()
         self._pool.close()
 
     def request_check(self) -> None:

pymongo/synchronous/topology.py

@@ -703,6 +703,8 @@ def close(self) -> None:
             # Stop SRV polling thread.
             if self._srv_monitor:
                 self._srv_monitor.close()
+                if not _IS_SYNC:
+                    self._srv_monitor.join()
 
             self._opened = False
             self._closed = True