class_docstr: A remote function.
  
  This is a decorated function. It can be used to spawn tasks.
  
  Attributes:
      _language: The target language.
      _function: The original function.
      _function_descriptor: The function descriptor. This is not defined
          until the remote function is first invoked because that is when the
          function is pickled, and the pickled function is used to compute
          the function descriptor.
      _function_name: The module and function name.
      _num_cpus: The default number of CPUs to use for invocations of this
          remote function.
      _num_gpus: The default number of GPUs to use for invocations of this
          remote function.
      _memory: The heap memory request in bytes for this task/actor,
          rounded down to the nearest integer.
      _resources: The default custom resource requirements for invocations of
          this remote function.
      _num_returns: The default number of return values for invocations
          of this remote function.
      _max_calls: The number of times a worker can execute this function
          before exiting.
      _max_retries: The number of times this task may be retried
          on worker failure.
      _retry_exceptions: Whether application-level errors should be retried.
          This can be a boolean or a list/tuple of exceptions that should be retried.
      _runtime_env: The runtime environment for this task.
      _decorator: An optional decorator that should be applied to the remote
          function invocation (as opposed to the function execution) before
          invoking the function. The decorator must return a function that
          takes in two arguments ("args" and "kwargs"). In most cases, it
          should call the function that was passed into the decorator and
          return the resulting ObjectRefs. For an example, see
          "test_decorated_function" in "python/ray/tests/test_basic.py".
      _function_signature: The function signature.
      _last_export_session_and_job: A pair of the last exported session
          and job to help us to know whether this function was exported.
          This is an imperfect mechanism used to determine if we need to
          export the remote function again. It is imperfect in the sense that
          the actor class definition could be exported multiple times by
          different workers.
      _scheduling_strategy: Strategy about how to schedule
          this remote function.
class_docstr_tok: ["A","remote","function",".","This","is","a","decorated","function",".","It","can","be","used","to","spawn","tasks",".","Attributes",":","_language",":","The","target","language",".","_function",":","The","original","function",".","_function_descriptor",":","The","function","descriptor",".","This","is","not","defined","until","the","remote","function","is","first","invoked","because","that","is","when","the","function","is","pickled",",","and","the","pickled","function","is","used","to","compute","the","function","descriptor",".","_function_name",":","The","module","and","function","name",".","_num_cpus",":","The","default","number","of","CPUs","to","use","for","invocations","of","this","remote","function",".","_num_gpus",":","The","default","number","of","GPUs","to","use","for","invocations","of","this","remote","function",".","_memory",":","The","heap","memory","request","in","bytes","for","this","task\/actor",",","rounded","down","to","the","nearest","integer",".","_resources",":","The","default","custom","resource","requirements","for","invocations","of","this","remote","function",".","_num_returns",":","The","default","number","of","return","values","for","invocations","of","this","remote","function",".","_max_calls",":","The","number","of","times","a","worker","can","execute","this","function","before","exiting",".","_max_retries",":","The","number","of","times","this","task","may","be","retried","on","worker","failure",".","_retry_exceptions",":","Whether","application-level","errors","should","be","retried",".","This","can","be","a","boolean","or","a","list\/tuple","of","exceptions","that","should","be","retried",".","_runtime_env",":","The","runtime","environment","for","this","task",".","_decorator",":","An","optional","decorator","that","should","be","applied","to","the","remote","function","invocation","(","as","opposed","to","the","function","execution",")","before","invoking","the","function",".","The","decorator","must","return","a","function","that","takes","in","two","arguments","(","``","args","''","and","``","kwargs","''",")",".","In","most","cases",",","it","should","call","the","function","that","was","passed","into","the","decorator","and","return","the","resulting","ObjectRefs",".","For","an","example",",","see","``","test_decorated_function","''","in","``","python\/ray\/tests\/test_basic.py","''",".","_function_signature",":","The","function","signature",".","_last_export_session_and_job",":","A","pair","of","the","last","exported","session","and","job","to","help","us","to","know","whether","this","function","was","exported",".","This","is","an","imperfect","mechanism","used","to","determine","if","we","need","to","export","the","remote","function","again",".","It","is","imperfect","in","the","sense","that","the","actor","class","definition","could","be","exported","multiple","times","by","different","workers",".","_scheduling_strategy",":","Strategy","about","how","to","schedule","this","remote","function","."]
docstr: Submit the remote function for execution.
docstr_tok: ["Submit","the","remote","function","for","execution","."]
code: def _remote(self, args=None, kwargs=None, **task_options):
      """Submit the remote function for execution."""
      # We pop the "max_calls" coming from "@ray.remote" here. We no longer need
      # it in "_remote()".
      task_options.pop("max_calls", None)
      if client_mode_should_convert():
          return client_mode_convert_function(
              self, args, kwargs, **task_options
          )
  
      worker = ray._private.worker.global_worker
      worker.check_connected()
  
      # We cannot do this when the function is first defined, because we need
      # ray.init() to have been called when this executes
      with self._inject_lock:
          if self._function_signature is None:
              self._function = _inject_tracing_into_function(
                  self._function
              )
              self._function_signature = (
                  ray._private.signature.extract_signature(
                      self._function
                  )
              )
  
      # If this function was not exported in this session and job, we need to
      # export this function again, because the current GCS doesn't have it.
      if (
          not self._is_cross_language
          and self._last_export_session_and_job
          != worker.current_session_and_job
      ):
          self._function_descriptor = (
              PythonFunctionDescriptor.from_function(
                  self._function, self._uuid
              )
          )
          # There is an interesting question here. If the remote function is
          # used by a subsequent driver (in the same script), should the
          # second driver pickle the function again? If yes, then the remote
          # function definition can differ in the second driver (e.g., if
          # variables in its closure have changed). We probably want the
          # behavior of the remote function in the second driver to be
          # independent of whether or not the function was invoked by the
          # first driver. This is an argument for repickling the function,
          # which we do here.
          self._pickled_function = pickle_dumps(
              self._function,
              f"Could not serialize the function {self._function_descriptor.repr}",
          )
  
          self._last_export_session_and_job = (
              worker.current_session_and_job
          )
          worker.function_actor_manager.export(self)
  
      kwargs = {} if kwargs is None else kwargs
      args = [] if args is None else args
  
      # fill task required options
      for k, v in ray_option_utils.task_options.items():
          if k == "max_retries":
              # TODO(swang): We need to override max_retries here because the default
              # value gets set at Ray import time. Ideally, we should allow setting
              # default values from env vars for other options too.
              v.default_value = os.environ.get(
                  "RAY_TASK_MAX_RETRIES", v.default_value
              )
              v.default_value = int(v.default_value)
          task_options[k] = task_options.get(k, v.default_value)
      # "max_calls" already takes effects and should not apply again.
      # Remove the default value here.
      task_options.pop("max_calls", None)
  
      # TODO(suquark): cleanup these fields
      name = task_options["name"]
      runtime_env = parse_runtime_env(task_options["runtime_env"])
      placement_group = task_options["placement_group"]
      placement_group_bundle_index = task_options[
          "placement_group_bundle_index"
      ]
      placement_group_capture_child_tasks = task_options[
          "placement_group_capture_child_tasks"
      ]
      scheduling_strategy = task_options["scheduling_strategy"]
  
      num_returns = task_options["num_returns"]
      if num_returns is None:
          if self._is_generator:
              num_returns = "streaming"
          else:
              num_returns = 1
  
      if num_returns == "dynamic":
          num_returns = -1
      elif num_returns == "streaming":
          # TODO(sang): This is a temporary private API.
          # Remove it when we migrate to the streaming generator.
          num_returns = ray._raylet.STREAMING_GENERATOR_RETURN
      generator_backpressure_num_objects = task_options[
          "_generator_backpressure_num_objects"
      ]
      if generator_backpressure_num_objects is None:
          generator_backpressure_num_objects = -1
  
      max_retries = task_options["max_retries"]
      retry_exceptions = task_options["retry_exceptions"]
      if isinstance(retry_exceptions, (list, tuple)):
          retry_exception_allowlist = tuple(retry_exceptions)
          retry_exceptions = True
      else:
          retry_exception_allowlist = None
  
      if scheduling_strategy is None or not isinstance(
          scheduling_strategy, PlacementGroupSchedulingStrategy
      ):
          _warn_if_using_deprecated_placement_group(task_options, 4)
  
      resources = ray._private.utils.resources_from_ray_options(
          task_options
      )
  
      if scheduling_strategy is None or isinstance(
          scheduling_strategy, PlacementGroupSchedulingStrategy
      ):
          if isinstance(
              scheduling_strategy, PlacementGroupSchedulingStrategy
          ):
              placement_group = scheduling_strategy.placement_group
              placement_group_bundle_index = (
                  scheduling_strategy.placement_group_bundle_index
              )
              placement_group_capture_child_tasks = (
                  scheduling_strategy.placement_group_capture_child_tasks
              )
  
          if placement_group_capture_child_tasks is None:
              placement_group_capture_child_tasks = (
                  worker.should_capture_child_tasks_in_placement_group
              )
          placement_group = _configure_placement_group_based_on_context(
              placement_group_capture_child_tasks,
              placement_group_bundle_index,
              resources,
              {},  # no placement_resources for tasks
              self._function_descriptor.function_name,
              placement_group=placement_group,
          )
          if not placement_group.is_empty:
              scheduling_strategy = PlacementGroupSchedulingStrategy(
                  placement_group,
                  placement_group_bundle_index,
                  placement_group_capture_child_tasks,
              )
          else:
              scheduling_strategy = "DEFAULT"
  
      serialized_runtime_env_info = None
      if runtime_env is not None:
          serialized_runtime_env_info = get_runtime_env_info(
              runtime_env,
              is_job_runtime_env=False,
              serialize=True,
          )
  
      if _task_launch_hook:
          _task_launch_hook(
              self._function_descriptor, resources, scheduling_strategy
          )
  
      def invocation(args, kwargs):
          if self._is_cross_language:
              list_args = cross_language._format_args(
                  worker, args, kwargs
              )
          elif not args and not kwargs and not self._function_signature:
              list_args = []
          else:
              list_args = ray._private.signature.flatten_args(
                  self._function_signature, args, kwargs
              )
  
          if worker.mode == ray._private.worker.LOCAL_MODE:
              assert (
                  not self._is_cross_language
              ), "Cross language remote function cannot be executed locally."
          object_refs = worker.core_worker.submit_task(
              self._language,
              self._function_descriptor,
              list_args,
              name if name is not None else "",
              num_returns,
              resources,
              max_retries,
              retry_exceptions,
              retry_exception_allowlist,
              scheduling_strategy,
              worker.debugger_breakpoint,
              serialized_runtime_env_info or "{}",
              generator_backpressure_num_objects,
          )
          # Reset worker's debug context from the last "remote" command
          # (which applies only to this .remote call).
          worker.debugger_breakpoint = b""
          if num_returns == STREAMING_GENERATOR_RETURN:
              # Streaming generator will return a single ref
              # that is for the generator task.
              assert len(object_refs) == 1
              generator_ref = object_refs[0]
              return ObjectRefGenerator(generator_ref, worker)
          if len(object_refs) == 1:
              return object_refs[0]
          elif len(object_refs) > 1:
              return object_refs
  
      if self._decorator is not None:
          invocation = self._decorator(invocation)
  
      return invocation(args, kwargs)
  
imports: import inspect
  import logging
  import os
  import uuid
  from functools import wraps
  from threading import Lock
  import ray._private.signature
  from ray import Language, cross_language
  from ray._private import ray_option_utils
  from ray._private.auto_init_hook import wrap_auto_init
  from ray._private.client_mode_hook import client_mode_convert_function, client_mode_should_convert
  from ray._private.ray_option_utils import _warn_if_using_deprecated_placement_group
  from ray._private.serialization import pickle_dumps
  from ray._private.utils import get_runtime_env_info, parse_runtime_env
  from ray._raylet import STREAMING_GENERATOR_RETURN, ObjectRefGenerator, PythonFunctionDescriptor
  from ray.util.annotations import DeveloperAPI, PublicAPI
  from ray.util.placement_group import _configure_placement_group_based_on_context
  from ray.util.scheduling_strategies import PlacementGroupSchedulingStrategy
  from ray.util.tracing.tracing_helper import _inject_tracing_into_function, _tracing_task_invocation
label_desc: Use image node_id 6 for calling the RemoteFunction obj's underlying member method code with example usage: obj._remote(args, kwargs) and returns: invocation, client_mode_convert_function, ObjectRefGenerator, object_refs, object_refs