class_docstr: A channel from ray driver/actor to ray actor. Use ray call reply for client channel recv.
class_docstr_tok: ["A","channel","from","ray","driver\/actor","to","ray","actor",".","Use","ray","call","reply","for","client","channel","recv","."]
code: def _submit_task(self, message: Any, object_ref: "ray.ObjectRef"):
      async def handle_task(message: Any, object_ref: "ray.ObjectRef"):
          # use `%.500` to avoid print too long messages
          with debug_async_timeout(
              "ray_object_retrieval_timeout",
              "Message that client sent to actor %s is %.500s and object_ref is %s",
              self.dest_address,
              message,
              object_ref,
          ):
              try:
                  result = await object_ref
              except Exception as e:  # pragma: no cover
                  # The error ClientObjectRef can't be formatted, so
                  # we give it a string `ClientObjectRef` instead.
                  try:
                      object_ref_str = str(object_ref)
                  except Exception:
                      object_ref_str = "ClientObjectRef"
                  logger.exception(
                      "Get object %s from %s failed, got exception %s.",
                      object_ref_str,
                      self.dest_address,
                      e,
                  )
                  raise
          if isinstance(result, RayChannelException):
              raise result.exc_value.with_traceback(
                  result.exc_traceback
              )
          return result.message
  
      def _on_completion(future):
          self._todo.remove(future)
          self._done.put_nowait(future)
  
      future = asyncio.ensure_future(handle_task(message, object_ref))
      future.add_done_callback(_on_completion)
      self._todo.add(future)
  
imports: import asyncio
  import concurrent.futures
  import itertools
  import logging
  import time
  from abc import ABC
  from collections import namedtuple
  from dataclasses import dataclass
  from typing import Any, Callable, Coroutine, Dict, List, Set, Tuple, Type
  from urllib.parse import urlparse
  from ....oscar.profiling import ProfilingData
  from ....serialization import serialize, deserialize
  from ....metrics import Metrics
  from ....utils import lazy_import, lazy_import_on_load, implements, classproperty, Timer
  from ...debug import debug_async_timeout
  from ...errors import ServerClosed
  from ..communication.base import Channel, ChannelType, Server, Client
  from ..communication.core import register_client, register_server
  from ..communication.errors import ChannelClosed
  from .utils import report_event
label_desc: Use image node_id 2 for calling the RayClientChannel obj's underlying member method code with example usage: obj._submit_task(message, object_ref) and returns: result