code: def _remote(
      self, args: Tuple[Any], kwargs: Dict[str, Any]
  ) -> concurrent.futures.Future:
      if not self.__class__ == DeploymentHandle:
          warnings.warn(
              "`DeploymentHandle` is now the default handle API. You can continue "
              "using the existing `RayServeHandle` and `RayServeSyncHandle` APIs "
              "by calling `handle.options(use_new_handle_api=False)` or setting the "
              "global environment variable `RAY_SERVE_ENABLE_NEW_HANDLE_API=0`, "
              "but support for these will be removed in a future release. "
              "See https://docs.ray.io/en/latest/serve/model_composition.html "
              "for more details."
          )
  
      self._record_telemetry_if_needed()
      _request_context = ray.serve.context._serve_request_context.get()
      request_metadata = RequestMetadata(
          _request_context.request_id,
          self.deployment_name,
          call_method=self.handle_options.method_name,
          route=_request_context.route,
          app_name=self.app_name,
          multiplexed_model_id=self.handle_options.multiplexed_model_id,
          is_streaming=self.handle_options.stream,
          _request_protocol=self.handle_options._request_protocol,
          grpc_context=_request_context.grpc_context,
      )
      self.request_counter.inc(
          tags={
              "route": _request_context.route,
              "application": _request_context.app_name,
          }
      )
      router, event_loop = self._get_or_create_router()
  
      # Schedule the coroutine to run on the router loop. This is always a separate
      # loop running in another thread to avoid user code blocking the router, so we
      # use the `concurrent.futures.Future` thread safe API.
      return asyncio.run_coroutine_threadsafe(
          router.assign_request(request_metadata, *args, **kwargs),
          loop=event_loop,
      )
  
imports: import asyncio
  import concurrent.futures
  import threading
  import warnings
  from dataclasses import dataclass
  from typing import Any, AsyncIterator, Dict, Iterator, Optional, Tuple, Union
  import ray
  from ray import serve
  from ray._raylet import GcsClient, ObjectRefGenerator
  from ray.serve._private.common import DeploymentID, RequestProtocol
  from ray.serve._private.default_impl import create_cluster_node_info_cache
  from ray.serve._private.router import RequestMetadata, Router
  from ray.serve._private.usage import ServeUsageTag
  from ray.serve._private.utils import DEFAULT, get_current_actor_id, get_random_string, is_running_in_asyncio_loop
  from ray.util import metrics
  from ray.util.annotations import Deprecated, DeveloperAPI, PublicAPI
label_desc: Use image node_id 10 for calling the _DeploymentHandleBase obj's underlying member method code with example usage: obj._remote(args, kwargs) and returns: asyncio