Add logs

python/xoscar/backends/communication/dummy.py

@@ -237,7 +237,12 @@ async def connect(
         client = DummyClient(local_address, dest_address, client_channel)
         client._task = task
         server._tasks.add(task)
-        task.add_done_callback(server._tasks.discard)
+
+        def _discard(t):
+            server._tasks.discard(t)
+            logger.info("Channel exit: %s", server_channel.info)
+
+        task.add_done_callback(_discard)
         return client
 
     @implements(Client.close)

python/xoscar/backends/communication/socket.py

@@ -179,6 +179,7 @@ async def on_connected(self, *args, **kwargs):
                 await channel.close()
             # Remove channel if channel exit
             self._channels.discard(channel)
+            logger.debug("Channel exit: %s", channel.info)
 
     @implements(Server.stop)
     async def stop(self):

python/xoscar/backends/communication/ucx.py

@@ -478,6 +478,7 @@ async def on_connected(self, *args, **kwargs):
                 await channel.close()
             # Remove channel if channel exit
             self._channels.discard(channel)
+            logger.debug("Channel exit: %s", channel.info)
 
     @implements(Server.stop)
     async def stop(self):

python/xoscar/backends/core.py

@@ -195,6 +195,7 @@ async def call(
         return await self.call_with_client(client, message, wait)
 
     async def stop(self):
+        logger.debug("Actor caller stop.")
         try:
             await asyncio.gather(*[client.close() for client in self._clients])
         except (ConnectionError, ServerClosed):
@@ -223,14 +224,18 @@ def __getattr__(self, item):
         try:
             actor_caller = self._thread_local.actor_caller
         except AttributeError:
+            thread_info = str(threading.current_thread())
+            logger.debug("Creating a new actor caller for thread: %s", thread_info)
             actor_caller = self._thread_local.actor_caller = ActorCallerThreadLocal()
             ref = self._thread_local.ref = ActorCaller._RefHolder()
             # If the thread exit, we clean the related actor callers and channels.
-            weakref.finalize(
-                ref,
-                asyncio.run_coroutine_threadsafe,
-                actor_caller.stop(),
-                self._close_loop,
-            )
+
+            def _cleanup():
+                asyncio.run_coroutine_threadsafe(actor_caller.stop(), self._close_loop)
+                logger.debug(
+                    "Clean up the actor caller due to thread exit: %s", thread_info
+                )
+
+            weakref.finalize(ref, _cleanup)
 
         return getattr(actor_caller, item)