Asyncio Module Compact Serialization [6/6] (#752)

Author: yuce

hazelcast/asyncio/client.py

@@ -8,15 +8,14 @@
 from hazelcast.config import Config, IndexConfig
 from hazelcast.internal.asyncio_connection import ConnectionManager, DefaultAddressProvider
 from hazelcast.core import DistributedObjectEvent, DistributedObjectInfo
-from hazelcast.cp import CPSubsystem, ProxySessionManager
 from hazelcast.discovery import HazelcastCloudAddressProvider
 from hazelcast.errors import IllegalStateError, InvalidConfigurationError
 from hazelcast.internal.asyncio_invocation import InvocationService, Invocation
 from hazelcast.internal.asyncio_proxy.vector_collection import VectorCollection
 from hazelcast.lifecycle import LifecycleService, LifecycleState, _InternalLifecycleService
 from hazelcast.internal.asyncio_listener import ClusterViewListenerService, ListenerService
 from hazelcast.near_cache import NearCacheManager
-from hazelcast.partition import PartitionService, _InternalPartitionService
+from hazelcast.internal.asyncio_partition import PartitionService, InternalPartitionService
 from hazelcast.protocol.codec import (
     client_add_distributed_object_listener_codec,
     client_get_distributed_objects_codec,
@@ -34,7 +33,7 @@
 from hazelcast.serialization import SerializationServiceV1
 from hazelcast.sql import SqlService, _InternalSqlService
 from hazelcast.internal.asyncio_statistics import Statistics
-from hazelcast.types import KeyType, ValueType, ItemType, MessageType
+from hazelcast.types import KeyType, ValueType
 from hazelcast.util import AtomicInteger, RoundRobinLB
 
 __all__ = ("HazelcastClient",)
@@ -84,7 +83,7 @@ def __init__(self, config: Config | None = None, **kwargs):
             self._config,
         )
         self._address_provider = self._create_address_provider()
-        self._internal_partition_service = _InternalPartitionService(self)
+        self._internal_partition_service = InternalPartitionService(self)
         self._partition_service = PartitionService(
             self._internal_partition_service,
             self._serialization_service,
@@ -111,8 +110,6 @@ def __init__(self, config: Config | None = None, **kwargs):
             self._compact_schema_service,
         )
         self._proxy_manager = ProxyManager(self._context)
-        self._cp_subsystem = CPSubsystem(self._context)
-        self._proxy_session_manager = ProxySessionManager(self._context)
         self._lock_reference_id_generator = AtomicInteger(1)
         self._statistics = Statistics(
             self,
@@ -159,15 +156,14 @@ def _init_context(self):
             self._near_cache_manager,
             self._lock_reference_id_generator,
             self._name,
-            self._proxy_session_manager,
             self._reactor,
             self._compact_schema_service,
         )
 
     async def _start(self):
         try:
             self._internal_lifecycle_service.start()
-            self._invocation_service.start()
+            await self._invocation_service.start()
             membership_listeners = self._config.membership_listeners
             self._internal_cluster_service.start(self._connection_manager, membership_listeners)
             self._cluster_view_listener.start()
@@ -278,7 +274,6 @@ async def shutdown(self) -> None:
             if self._internal_lifecycle_service.running:
                 self._internal_lifecycle_service.fire_lifecycle_event(LifecycleState.SHUTTING_DOWN)
                 self._internal_lifecycle_service.shutdown()
-                self._proxy_session_manager.shutdown().result()
                 self._near_cache_manager.destroy_near_caches()
                 await self._connection_manager.shutdown()
                 self._invocation_service.shutdown()
@@ -301,10 +296,6 @@ def partition_service(self) -> PartitionService:
     def cluster_service(self) -> ClusterService:
         return self._cluster_service
 
-    @property
-    def cp_subsystem(self) -> CPSubsystem:
-        return self._cp_subsystem
-
     def _create_address_provider(self):
         config = self._config
         cluster_members = config.cluster_members
@@ -360,7 +351,6 @@ def __init__(self):
         self.near_cache_manager = None
         self.lock_reference_id_generator = None
         self.name = None
-        self.proxy_session_manager = None
         self.reactor = None
         self.compact_schema_service = None
 
@@ -378,7 +368,6 @@ def init_context(
         near_cache_manager,
         lock_reference_id_generator,
         name,
-        proxy_session_manager,
         reactor,
         compact_schema_service,
     ):
@@ -394,6 +383,5 @@ def init_context(
         self.near_cache_manager = near_cache_manager
         self.lock_reference_id_generator = lock_reference_id_generator
         self.name = name
-        self.proxy_session_manager = proxy_session_manager
         self.reactor = reactor
         self.compact_schema_service = compact_schema_service

hazelcast/internal/asyncio_compact.py

@@ -57,77 +57,64 @@ def fetch_schema(self, schema_id: int) -> asyncio.Future:
         self._invocation_service.invoke(fetch_schema_invocation)
         return fetch_schema_invocation.future
 
-    def send_schema_and_retry(
+    async def send_schema_and_retry(
         self,
         error: "SchemaNotReplicatedError",
         func: typing.Callable[..., asyncio.Future],
         *args: typing.Any,
         **kwargs: typing.Any,
-    ) -> asyncio.Future:
+    ) -> None:
         schema = error.schema
         clazz = error.clazz
         request = client_send_schema_codec.encode_request(schema)
 
-        def callback():
+        async def callback():
             self._has_replicated_schemas = True
             self._compact_serializer.register_schema_to_type(schema, clazz)
-            return func(*args, **kwargs)
+            maybe_coro = func(*args, **kwargs)
+            # maybe_coro maybe a coroutine or None
+            if maybe_coro:
+                return await maybe_coro
 
-        return self._replicate_schema(
-            schema, request, CompactSchemaService._SEND_SCHEMA_RETRY_COUNT, callback
+        return await self._replicate_schema(
+            schema, request, CompactSchemaService._SEND_SCHEMA_RETRY_COUNT, callback()
         )
 
-    def _replicate_schema(
+    async def _replicate_schema(
         self,
         schema: "Schema",
         request: "OutboundMessage",
         remaining_retries: int,
-        callback: typing.Callable[..., asyncio.Future],
-    ) -> asyncio.Future:
-        def continuation(future: asyncio.Future):
-            replicated_members = future.result()
+        callback: typing.Coroutine[typing.Any, typing.Any, typing.Any],
+    ) -> None:
+        while remaining_retries >= 2:
+            replicated_members = await self._send_schema_replication_request(request)
             members = self._cluster_service.get_members()
             for member in members:
                 if member.uuid not in replicated_members:
                     break
             else:
                 # Loop completed normally.
                 # All members in our member list all known to have the schema
-                return callback()
+                return await callback
 
             # There is a member in our member list that the schema
             # is not known to be replicated yet. We should retry
             # sending it in a random member.
-            if remaining_retries <= 1:
-                # We tried to send it a couple of times, but the member list
-                # in our local and the member list returned by the initiator
-                # nodes did not match.
-                raise IllegalStateError(
-                    f"The schema {schema} cannot be replicated in the cluster, "
-                    f"after {CompactSchemaService._SEND_SCHEMA_RETRY_COUNT} retries. "
-                    f"It might be the case that the client is connected to the two "
-                    f"halves of the cluster that is experiencing a split-brain, "
-                    f"and continue putting the data associated with that schema "
-                    f"might result in data loss. It might be possible to replicate "
-                    f"the schema after some time, when the cluster is healed."
-                )
-
-            delayed_future: asyncio.Future = asyncio.get_running_loop().create_future()
-            self._reactor.add_timer(
-                self._invocation_retry_pause,
-                lambda: delayed_future.set_result(None),
-            )
-
-            def retry(_):
-                return self._replicate_schema(
-                    schema, request.copy(), remaining_retries - 1, callback
-                )
-
-            return delayed_future.add_done_callback(retry)
-
-        fut = self._send_schema_replication_request(request)
-        fut.add_done_callback(continuation)
-        return fut
+            await asyncio.sleep(self._invocation_retry_pause)
+
+        # We tried to send it a couple of times, but the member list
+        # in our local and the member list returned by the initiator
+        # nodes did not match.
+        raise IllegalStateError(
+            f"The schema {schema} cannot be replicated in the cluster, "
+            f"after {CompactSchemaService._SEND_SCHEMA_RETRY_COUNT} retries. "
+            f"It might be the case that the client is connected to the two "
+            f"halves of the cluster that is experiencing a split-brain, "
+            f"and continue putting the data associated with that schema "
+            f"might result in data loss. It might be possible to replicate "
+            f"the schema after some time, when the cluster is healed."
+        )
 
     def _send_schema_replication_request(self, request: "OutboundMessage") -> asyncio.Future:
         invocation = Invocation(request, response_handler=client_send_schema_codec.decode_response)

hazelcast/internal/asyncio_invocation.py

@@ -96,7 +96,7 @@ def __init__(self, client, config, reactor):
         self._backup_ack_to_client_enabled = smart_routing and config.backup_ack_to_client_enabled
         self._fail_on_indeterminate_state = config.fail_on_indeterminate_operation_state
         self._backup_timeout = config.operation_backup_timeout
-        self._clean_resources_timer = None
+        self._clean_resources_task = None
         self._shutdown = False
         self._compact_schema_service = None
 
@@ -107,8 +107,8 @@ def init(self, partition_service, connection_manager, listener_service, compact_
         self._check_invocation_allowed_fn = connection_manager.check_invocation_allowed
         self._compact_schema_service = compact_schema_service
 
-    def start(self):
-        self._start_clean_resources_timer()
+    async def start(self):
+        await self._start_clean_resources_timer()
 
     async def add_backup_listener(self):
         if self._backup_ack_to_client_enabled:
@@ -152,8 +152,8 @@ def shutdown(self):
             return
 
         self._shutdown = True
-        if self._clean_resources_timer:
-            self._clean_resources_timer.cancel()
+        if self._clean_resources_task:
+            self._clean_resources_task.cancel()
         for invocation in list(self._pending.values()):
             self._notify_error(invocation, HazelcastClientNotActiveError())
 
@@ -400,8 +400,9 @@ def _notify_backup_complete(self, invocation):
 
         self._complete(invocation, invocation.pending_response)
 
-    def _start_clean_resources_timer(self):
-        def run():
+    async def _start_clean_resources_timer(self):
+        async def run():
+            await asyncio.sleep(self._CLEAN_RESOURCES_PERIOD)
             if self._shutdown:
                 return
 
@@ -419,9 +420,9 @@ def run():
                 if self._backup_ack_to_client_enabled:
                     self._detect_and_handle_backup_timeout(invocation, now)
 
-            self._clean_resources_timer = self._reactor.add_timer(self._CLEAN_RESOURCES_PERIOD, run)
+            self._clean_resources_task = asyncio.create_task(run())
 
-        self._clean_resources_timer = self._reactor.add_timer(self._CLEAN_RESOURCES_PERIOD, run)
+        self._clean_resources_task = asyncio.create_task(run())
 
     def _detect_and_handle_backup_timeout(self, invocation, now):
         if not invocation.pending_response: