Updates

Author: yuce

hazelcast/internal/asyncio_connection.py

@@ -173,7 +173,7 @@ def __init__(
             self, self._client, config, reactor, invocation_service
         )
         self._connection_listeners = []
-        self._connect_all_members_timer = None
+        self._connect_all_members_task: asyncio.Task | None = None
         self._async_start = config.async_start
         self._connect_to_cluster_thread_running = False
         self._shuffle_member_list = config.shuffle_member_list
@@ -274,8 +274,8 @@ async def shutdown(self):
             return
 
         self.live = False
-        if self._connect_all_members_timer:
-            self._connect_all_members_timer.cancel()
+        if self._connect_all_members_task:
+            self._connect_all_members_task.cancel()
 
         self._heartbeat_manager.shutdown()
 
@@ -458,6 +458,7 @@ def _start_connect_all_members_timer(self):
         connecting_uuids = set()
 
         async def run():
+            await asyncio.sleep(1)
             if not self._lifecycle_service.running:
                 return
 
@@ -479,13 +480,9 @@ async def run():
             for item in member_uuids:
                 connecting_uuids.discard(item)
 
-            self._connect_all_members_timer = self._reactor.add_timer(
-                1, lambda: asyncio.create_task(run())
-            )
+            self._connect_all_members_task = asyncio.create_task(run())
 
-        self._connect_all_members_timer = self._reactor.add_timer(
-            1, lambda: asyncio.create_task(run())
-        )
+        self._connect_all_members_task = asyncio.create_task(run())
 
     async def _connect_to_cluster(self):
         await self._sync_connect_to_cluster()
@@ -816,12 +813,14 @@ def __init__(self, connection_manager, client, config, reactor, invocation_servi
         self._invocation_service = invocation_service
         self._heartbeat_timeout = config.heartbeat_timeout
         self._heartbeat_interval = config.heartbeat_interval
-        self._heartbeat_timer = None
+        self._heartbeat_task: asyncio.Task | None = None
 
     def start(self):
         """Starts sending periodic HeartBeat operations."""
 
         async def _heartbeat():
+            await asyncio.sleep(self._heartbeat_interval)
+            _logger.debug("heartbeat")
             conn_manager = self._connection_manager
             if not conn_manager.live:
                 return
@@ -830,18 +829,14 @@ async def _heartbeat():
             async with asyncio.TaskGroup() as tg:
                 for connection in list(conn_manager.active_connections.values()):
                     tg.create_task(self._check_connection(now, connection))
-            self._heartbeat_timer = self._reactor.add_timer(
-                self._heartbeat_interval, lambda: asyncio.create_task(_heartbeat())
-            )
+            self._heartbeat_task = asyncio.create_task(_heartbeat())
 
-        self._heartbeat_timer = self._reactor.add_timer(
-            self._heartbeat_interval, lambda: asyncio.create_task(_heartbeat())
-        )
+        self._heartbeat_task = asyncio.create_task(_heartbeat())
 
     def shutdown(self):
         """Stops HeartBeat operations."""
-        if self._heartbeat_timer:
-            self._heartbeat_timer.cancel()
+        if self._heartbeat_task:
+            self._heartbeat_task.cancel()
 
     async def _check_connection(self, now, connection):
         if not connection.live:
@@ -858,7 +853,7 @@ async def _check_connection(self, now, connection):
         if (now - connection.last_write_time) > self._heartbeat_interval:
             request = client_ping_codec.encode_request()
             invocation = Invocation(request, connection=connection, urgent=True)
-            self._invocation_service.invoke(invocation)
+            asyncio.create_task(self._invocation_service.ainvoke(invocation))
 
 
 _frame_header = struct.Struct("<iH")

hazelcast/internal/asyncio_reactor.py

@@ -17,7 +17,6 @@ class AsyncioReactor:
     def __init__(self, loop: AbstractEventLoop | None = None):
         self._is_live = False
         self._loop = loop or asyncio.get_running_loop()
-        self._bytes_lock = asyncio.Lock()
         self._bytes_sent = 0
         self._bytes_received = 0
 
@@ -46,17 +45,11 @@ async def connection_factory(
         )
 
     def update_bytes_sent(self, sent: int):
-        # with self._bytes_lock:
         self._bytes_sent += sent
 
     def update_bytes_received(self, received: int):
-        # with self._bytes_lock:
         self._bytes_received += received
 
-    # def _asyncio_loop(self):
-    #     asyncio.set_event_loop(self._loop)
-    #     self._loop.run_forever()
-
 
 class AsyncioConnection(Connection):
     def __init__(