Only join executors on async

Author: NoahStapp

pymongo/asynchronous/mongo_client.py

@@ -1559,14 +1559,14 @@ 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()
         self._closed = True
-        # Yield to the asyncio event loop so all executor tasks properly exit after cancellation
-        await asyncio.sleep(0)
 
     if not _IS_SYNC:
         # Add support for contextlib.aclosing.

pymongo/asynchronous/monitor.py

@@ -191,6 +191,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.
@@ -458,6 +460,8 @@ def __init__(self, topology: Topology, topology_settings: TopologySettings, pool
 
     async def close(self) -> None:
         self.gc_safe_close()
+        if not _IS_SYNC:
+            await self._executor.join()
         # Increment the generation and maybe close the socket. If the executor
         # thread has the socket checked out, it will be closed when checked in.
         await self._pool.reset()

pymongo/periodic_executor.py

@@ -75,19 +75,12 @@ def close(self, dummy: Any = None) -> None:
         callback; see monitor.py.
         """
         self._stopped = True
-        if self._task:
+        if self._task is not None:
             self._task.cancel()
 
     async def join(self, timeout: Optional[int] = None) -> None:
         if self._task is not None:
-            try:
-                await asyncio.wait_for(self._task, timeout=timeout)  # type-ignore: [arg-type]
-            except asyncio.TimeoutError:
-                # Task timed out
-                pass
-            except asyncio.exceptions.CancelledError:
-                # Task was already finished, or not yet started.
-                raise
+            await asyncio.wait([self._task], timeout=timeout)  # type-ignore: [arg-type]
 
     def wake(self) -> None:
         """Execute the target function soon."""

pymongo/synchronous/mongo_client.py

@@ -1553,13 +1553,14 @@ 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()
         self._closed = True
-        # Yield to the asyncio event loop so all executor tasks properly exit after cancellation
 
     if not _IS_SYNC:
         # Add support for contextlib.closing.

pymongo/synchronous/monitor.py

@@ -191,6 +191,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.
@@ -458,6 +460,8 @@ def __init__(self, topology: Topology, topology_settings: TopologySettings, pool
 
     def close(self) -> None:
         self.gc_safe_close()
+        if not _IS_SYNC:
+            self._executor.join()
         # Increment the generation and maybe close the socket. If the executor
         # thread has the socket checked out, it will be closed when checked in.
         self._pool.reset()

test/__init__.py

@@ -1136,8 +1136,6 @@ class IntegrationTest(PyMongoTestCase):
 
     @client_context.require_connection
     def setUp(self) -> None:
-        if not _IS_SYNC:
-            reset_client_context()
         if client_context.load_balancer and not getattr(self, "RUN_ON_LOAD_BALANCER", False):
             raise SkipTest("this test does not support load balancers")
         if client_context.serverless and not getattr(self, "RUN_ON_SERVERLESS", False):

test/asynchronous/__init__.py

@@ -1154,8 +1154,6 @@ class AsyncIntegrationTest(AsyncPyMongoTestCase):
 
     @async_client_context.require_connection
     async def asyncSetUp(self) -> None:
-        if not _IS_SYNC:
-            await reset_client_context()
         if async_client_context.load_balancer and not getattr(self, "RUN_ON_LOAD_BALANCER", False):
             raise SkipTest("this test does not support load balancers")
         if async_client_context.serverless and not getattr(self, "RUN_ON_SERVERLESS", False):