PYTHON-4725 Async client should use tasks for SDAM instead of threads

pymongo/asynchronous/mongo_client.py

@@ -59,8 +59,8 @@
 
 from bson.codec_options import DEFAULT_CODEC_OPTIONS, CodecOptions, TypeRegistry
 from bson.timestamp import Timestamp
-from pymongo import _csot, common, helpers_shared, uri_parser
-from pymongo.asynchronous import client_session, database, periodic_executor
+from pymongo import _csot, common, helpers_shared, periodic_executor, uri_parser
+from pymongo.asynchronous import client_session, database
 from pymongo.asynchronous.change_stream import AsyncChangeStream, AsyncClusterChangeStream
 from pymongo.asynchronous.client_bulk import _AsyncClientBulk
 from pymongo.asynchronous.client_session import _EmptyServerSession
@@ -908,7 +908,7 @@ async def target() -> bool:
             await AsyncMongoClient._process_periodic_tasks(client)
             return True
 
-        executor = periodic_executor.PeriodicExecutor(
+        executor = periodic_executor.AsyncPeriodicExecutor(
             interval=common.KILL_CURSOR_FREQUENCY,
             min_interval=common.MIN_HEARTBEAT_INTERVAL,
             target=target,

pymongo/asynchronous/monitor.py

@@ -22,14 +22,13 @@
 import weakref
 from typing import TYPE_CHECKING, Any, Mapping, Optional, cast
 
-from pymongo import common
+from pymongo import common, periodic_executor
 from pymongo._csot import MovingMinimum
-from pymongo.asynchronous import periodic_executor
-from pymongo.asynchronous.periodic_executor import _shutdown_executors
 from pymongo.errors import NetworkTimeout, NotPrimaryError, OperationFailure, _OperationCancelled
 from pymongo.hello import Hello
 from pymongo.lock import _create_lock
 from pymongo.logger import _SDAM_LOGGER, _debug_log, _SDAMStatusMessage
+from pymongo.periodic_executor import _shutdown_executors
 from pymongo.pool_options import _is_faas
 from pymongo.read_preferences import MovingAverage
 from pymongo.server_description import ServerDescription
@@ -76,7 +75,7 @@ async def target() -> bool:
             await monitor._run()  # type:ignore[attr-defined]
             return True
 
-        executor = periodic_executor.PeriodicExecutor(
+        executor = periodic_executor.AsyncPeriodicExecutor(
             interval=interval, min_interval=min_interval, target=target, name=name
         )
 
@@ -112,9 +111,9 @@ async def close(self) -> None:
         """
         self.gc_safe_close()
 
-    def join(self, timeout: Optional[int] = None) -> None:
+    async def join(self, timeout: Optional[int] = None) -> None:
         """Wait for the monitor to stop."""
-        self._executor.join(timeout)
+        await self._executor.join(timeout)
 
     def request_check(self) -> None:
         """If the monitor is sleeping, wake it soon."""
@@ -139,7 +138,7 @@ def __init__(
         """
         super().__init__(
             topology,
-            "pymongo_server_monitor_thread",
+            "pymongo_server_monitor_task",
             topology_settings.heartbeat_frequency,
             common.MIN_HEARTBEAT_INTERVAL,
         )
@@ -250,7 +249,7 @@ async def _check_server(self) -> ServerDescription:
             except (OperationFailure, NotPrimaryError) as exc:
                 # Update max cluster time even when hello fails.
                 details = cast(Mapping[str, Any], exc.details)
-                self._topology.receive_cluster_time(details.get("$clusterTime"))
+                await self._topology.receive_cluster_time(details.get("$clusterTime"))
                 raise
         except ReferenceError:
             raise
@@ -434,7 +433,7 @@ def __init__(self, topology: Topology, topology_settings: TopologySettings, pool
         """
         super().__init__(
             topology,
-            "pymongo_server_rtt_thread",
+            "pymongo_server_rtt_task",
             topology_settings.heartbeat_frequency,
             common.MIN_HEARTBEAT_INTERVAL,
         )
@@ -531,4 +530,5 @@ def _shutdown_resources() -> None:
         shutdown()
 
 
-atexit.register(_shutdown_resources)
+if _IS_SYNC:
+    atexit.register(_shutdown_resources)

pymongo/asynchronous/topology.py

@@ -27,8 +27,7 @@
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, Callable, Mapping, Optional, cast
 
-from pymongo import _csot, common, helpers_shared
-from pymongo.asynchronous import periodic_executor
+from pymongo import _csot, common, helpers_shared, periodic_executor
 from pymongo.asynchronous.client_session import _ServerSession, _ServerSessionPool
 from pymongo.asynchronous.monitor import SrvMonitor
 from pymongo.asynchronous.pool import Pool
@@ -185,7 +184,7 @@ def __init__(self, topology_settings: TopologySettings):
             async def target() -> bool:
                 return process_events_queue(weak)
 
-            executor = periodic_executor.PeriodicExecutor(
+            executor = periodic_executor.AsyncPeriodicExecutor(
                 interval=common.EVENTS_QUEUE_FREQUENCY,
                 min_interval=common.MIN_HEARTBEAT_INTERVAL,
                 target=target,
@@ -742,7 +741,7 @@ async def close(self) -> None:
         if self._publish_server or self._publish_tp:
             # Make sure the events executor thread is fully closed before publishing the remaining events
             self.__events_executor.close()
-            self.__events_executor.join(1)
+            await self.__events_executor.join(1)
             process_events_queue(weakref.ref(self._events))  # type: ignore[arg-type]
 
     @property