Address review

Author: NoahStapp

pymongo/asynchronous/mongo_client.py

@@ -1566,8 +1566,13 @@ async def close(self) -> None:
             await self._encrypter.close()
         self._closed = True
         if not _IS_SYNC:
-            self._topology._monitor_tasks.append(self._kill_cursors_executor)  # type: ignore[arg-type]
-            join_tasks = [t.join() for t in self._topology._monitor_tasks]  # type: ignore[func-returns-value]
+            join_tasks = [self._kill_cursors_executor]
+            try:
+                while self._topology._monitor_tasks:
+                    join_tasks.append(self._topology._monitor_tasks.pop())
+            except IndexError:
+                pass
+            join_tasks = [t.join() for t in join_tasks]  # type: ignore[func-returns-value]
             await asyncio.gather(*join_tasks)
 
     if not _IS_SYNC:

pymongo/asynchronous/monitor.py

@@ -190,8 +190,7 @@ def gc_safe_close(self) -> None:
         self.cancel_check()
 
     async def join(self) -> None:
-        await self._executor.join()
-        await self._rtt_monitor.join()
+        await asyncio.gather(*[self._executor.join(), self._rtt_monitor.join()])
 
     async def close(self) -> None:
         self.gc_safe_close()

pymongo/asynchronous/topology.py

@@ -245,7 +245,8 @@ async def open(self) -> None:
                 # Close servers and clear the pools.
                 for server in self._servers.values():
                     await server.close()
-                    self._monitor_tasks.append(server._monitor)
+                    if not _IS_SYNC:
+                        self._monitor_tasks.append(server._monitor)
                 # Reset the session pool to avoid duplicate sessions in
                 # the child process.
                 self._session_pool.reset()
@@ -298,9 +299,14 @@ async def select_servers(
             ]
 
         if not _IS_SYNC and self._monitor_tasks:
-            joins = [t.join() for t in self._monitor_tasks]  # type: ignore[func-returns-value]
-            await asyncio.gather(*joins)
-            self._monitor_tasks = []
+            join_tasks = []
+            try:
+                while self._monitor_tasks:
+                    join_tasks.append(self._monitor_tasks.pop())
+            except IndexError:
+                pass
+            join_tasks = [t.join() for t in join_tasks]  # type: ignore[func-returns-value]
+            await asyncio.gather(*join_tasks)
 
         return servers
 
@@ -532,7 +538,8 @@ async def _process_change(
             and self._description.topology_type not in SRV_POLLING_TOPOLOGIES
         ):
             await self._srv_monitor.close()
-            self._monitor_tasks.append(self._srv_monitor)
+            if not _IS_SYNC:
+                self._monitor_tasks.append(self._srv_monitor)
 
         # Clear the pool from a failed heartbeat.
         if reset_pool:
@@ -708,7 +715,8 @@ async def close(self) -> None:
             old_td = self._description
             for server in self._servers.values():
                 await server.close()
-                self._monitor_tasks.append(server._monitor)
+                if not _IS_SYNC:
+                    self._monitor_tasks.append(server._monitor)
 
             # Mark all servers Unknown.
             self._description = self._description.reset()
@@ -719,7 +727,8 @@ async def close(self) -> None:
             # Stop SRV polling thread.
             if self._srv_monitor:
                 await self._srv_monitor.close()
-                self._monitor_tasks.append(self._srv_monitor)
+                if not _IS_SYNC:
+                    self._monitor_tasks.append(self._srv_monitor)
 
             self._opened = False
             self._closed = True
@@ -959,7 +968,8 @@ async def _update_servers(self) -> None:
         for address, server in list(self._servers.items()):
             if not self._description.has_server(address):
                 await server.close()
-                self._monitor_tasks.append(server._monitor)
+                if not _IS_SYNC:
+                    self._monitor_tasks.append(server._monitor)
                 self._servers.pop(address)
 
     def _create_pool_for_server(self, address: _Address) -> Pool:

pymongo/synchronous/mongo_client.py

@@ -1560,8 +1560,13 @@ def close(self) -> None:
             self._encrypter.close()
         self._closed = True
         if not _IS_SYNC:
-            self._topology._monitor_tasks.append(self._kill_cursors_executor)  # type: ignore[arg-type]
-            join_tasks = [t.join() for t in self._topology._monitor_tasks]  # type: ignore[func-returns-value]
+            join_tasks = [self._kill_cursors_executor]
+            try:
+                while self._topology._monitor_tasks:
+                    join_tasks.append(self._topology._monitor_tasks.pop())
+            except IndexError:
+                pass
+            join_tasks = [t.join() for t in join_tasks]  # type: ignore[func-returns-value]
             asyncio.gather(*join_tasks)
 
     if not _IS_SYNC:

pymongo/synchronous/monitor.py

@@ -190,8 +190,7 @@ def gc_safe_close(self) -> None:
         self.cancel_check()
 
     def join(self) -> None:
-        self._executor.join()
-        self._rtt_monitor.join()
+        asyncio.gather(*[self._executor.join(), self._rtt_monitor.join()])
 
     def close(self) -> None:
         self.gc_safe_close()

pymongo/synchronous/topology.py

@@ -245,7 +245,8 @@ def open(self) -> None:
                 # Close servers and clear the pools.
                 for server in self._servers.values():
                     server.close()
-                    self._monitor_tasks.append(server._monitor)
+                    if not _IS_SYNC:
+                        self._monitor_tasks.append(server._monitor)
                 # Reset the session pool to avoid duplicate sessions in
                 # the child process.
                 self._session_pool.reset()
@@ -298,9 +299,14 @@ def select_servers(
             ]
 
         if not _IS_SYNC and self._monitor_tasks:
-            joins = [t.join() for t in self._monitor_tasks]  # type: ignore[func-returns-value]
-            asyncio.gather(*joins)
-            self._monitor_tasks = []
+            join_tasks = []
+            try:
+                while self._monitor_tasks:
+                    join_tasks.append(self._monitor_tasks.pop())
+            except IndexError:
+                pass
+            join_tasks = [t.join() for t in join_tasks]  # type: ignore[func-returns-value]
+            asyncio.gather(*join_tasks)
 
         return servers
 
@@ -532,7 +538,8 @@ def _process_change(
             and self._description.topology_type not in SRV_POLLING_TOPOLOGIES
         ):
             self._srv_monitor.close()
-            self._monitor_tasks.append(self._srv_monitor)
+            if not _IS_SYNC:
+                self._monitor_tasks.append(self._srv_monitor)
 
         # Clear the pool from a failed heartbeat.
         if reset_pool:
@@ -706,7 +713,8 @@ def close(self) -> None:
             old_td = self._description
             for server in self._servers.values():
                 server.close()
-                self._monitor_tasks.append(server._monitor)
+                if not _IS_SYNC:
+                    self._monitor_tasks.append(server._monitor)
 
             # Mark all servers Unknown.
             self._description = self._description.reset()
@@ -717,7 +725,8 @@ def close(self) -> None:
             # Stop SRV polling thread.
             if self._srv_monitor:
                 self._srv_monitor.close()
-                self._monitor_tasks.append(self._srv_monitor)
+                if not _IS_SYNC:
+                    self._monitor_tasks.append(self._srv_monitor)
 
             self._opened = False
             self._closed = True
@@ -957,7 +966,8 @@ def _update_servers(self) -> None:
         for address, server in list(self._servers.items()):
             if not self._description.has_server(address):
                 server.close()
-                self._monitor_tasks.append(server._monitor)
+                if not _IS_SYNC:
+                    self._monitor_tasks.append(server._monitor)
                 self._servers.pop(address)
 
     def _create_pool_for_server(self, address: _Address) -> Pool: