class_docstr: An actor class.
  
  This is a decorated class. It can be used to create actors.
  
  Attributes:
      __ray_metadata__: Contains metadata for the actor.
class_docstr_tok: ["An","actor","class",".","This","is","a","decorated","class",".","It","can","be","used","to","create","actors",".","Attributes",":","__ray_metadata__",":","Contains","metadata","for","the","actor","."]
docstr: Create an actor.
  
  This method allows more flexibility than the remote method because
  resource requirements can be specified and override the defaults in the
  decorator.
  
  Args:
      args: The arguments to forward to the actor constructor.
      kwargs: The keyword arguments to forward to the actor constructor.
      num_cpus: The number of CPUs required by the actor creation task.
      num_gpus: The number of GPUs required by the actor creation task.
      memory: Restrict the heap memory usage of this actor.
      resources: The custom resources required by the actor creation
          task.
      max_concurrency: The max number of concurrent calls to allow for
          this actor. This only works with direct actor calls. The max
          concurrency defaults to 1 for threaded execution, and 1000 for
          asyncio execution. Note that the execution order is not
          guaranteed when max_concurrency > 1.
      name: The globally unique name for the actor, which can be used
          to retrieve the actor via ray.get_actor(name) as long as the
          actor is still alive.
      namespace: Override the namespace to use for the actor. By default,
          actors are created in an anonymous namespace. The actor can
          be retrieved via ray.get_actor(name=name, namespace=namespace).
      lifetime: Either `None`, which defaults to the actor will fate
          share with its creator and will be deleted once its refcount
          drops to zero, or "detached", which means the actor will live
          as a global object independent of the creator.
      placement_group: (This has been deprecated, please use
          `PlacementGroupSchedulingStrategy` scheduling_strategy)
          the placement group this actor belongs to,
          or None if it doesn't belong to any group. Setting to "default"
          autodetects the placement group based on the current setting of
          placement_group_capture_child_tasks.
      placement_group_bundle_index: (This has been deprecated, please use
          `PlacementGroupSchedulingStrategy` scheduling_strategy)
          the index of the bundle
          if the actor belongs to a placement group, which may be -1 to
          specify any available bundle.
      placement_group_capture_child_tasks: (This has been deprecated,
          please use `PlacementGroupSchedulingStrategy`
          scheduling_strategy)
          Whether or not children tasks
          of this actor should implicitly use the same placement group
          as its parent. It is False by default.
      runtime_env (Dict[str, Any]): Specifies the runtime environment for
          this actor or task and its children (see
          :ref:`runtime-environments` for details).
      max_pending_calls: Set the max number of pending calls
          allowed on the actor handle. When this value is exceeded,
          PendingCallsLimitExceeded will be raised for further tasks.
          Note that this limit is counted per handle. -1 means that the
          number of pending calls is unlimited.
      scheduling_strategy: Strategy about how to schedule this actor.
  
  Returns:
      A handle to the newly created actor.
docstr_tok: ["Create","an","actor",".","This","method","allows","more","flexibility","than","the","remote","method","because","resource","requirements","can","be","specified","and","override","the","defaults","in","the","decorator",".","Args",":","args",":","The","arguments","to","forward","to","the","actor","constructor",".","kwargs",":","The","keyword","arguments","to","forward","to","the","actor","constructor",".","num_cpus",":","The","number","of","CPUs","required","by","the","actor","creation","task",".","num_gpus",":","The","number","of","GPUs","required","by","the","actor","creation","task",".","memory",":","Restrict","the","heap","memory","usage","of","this","actor",".","resources",":","The","custom","resources","required","by","the","actor","creation","task",".","max_concurrency",":","The","max","number","of","concurrent","calls","to","allow","for","this","actor",".","This","only","works","with","direct","actor","calls",".","The","max","concurrency","defaults","to","1","for","threaded","execution",",","and","1000","for","asyncio","execution",".","Note","that","the","execution","order","is","not","guaranteed","when","max_concurrency",">","1.","name",":","The","globally","unique","name","for","the","actor",",","which","can","be","used","to","retrieve","the","actor","via","ray.get_actor","(","name",")","as","long","as","the","actor","is","still","alive",".","namespace",":","Override","the","namespace","to","use","for","the","actor",".","By","default",",","actors","are","created","in","an","anonymous","namespace",".","The","actor","can","be","retrieved","via","ray.get_actor","(","name=name",",","namespace=namespace",")",".","lifetime",":","Either","`","None","`",",","which","defaults","to","the","actor","will","fate","share","with","its","creator","and","will","be","deleted","once","its","refcount","drops","to","zero",",","or","``","detached","''",",","which","means","the","actor","will","live","as","a","global","object","independent","of","the","creator",".","placement_group",":","(","This","has","been","deprecated",",","please","use","`","PlacementGroupSchedulingStrategy","`","scheduling_strategy",")","the","placement","group","this","actor","belongs","to",",","or","None","if","it","does","n't","belong","to","any","group",".","Setting","to","``","default","''","autodetects","the","placement","group","based","on","the","current","setting","of","placement_group_capture_child_tasks",".","placement_group_bundle_index",":","(","This","has","been","deprecated",",","please","use","`","PlacementGroupSchedulingStrategy","`","scheduling_strategy",")","the","index","of","the","bundle","if","the","actor","belongs","to","a","placement","group",",","which","may","be","-1","to","specify","any","available","bundle",".","placement_group_capture_child_tasks",":","(","This","has","been","deprecated",",","please","use","`","PlacementGroupSchedulingStrategy","`","scheduling_strategy",")","Whether","or","not","children","tasks","of","this","actor","should","implicitly","use","the","same","placement","group","as","its","parent",".","It","is","False","by","default",".","runtime_env","(","Dict","[","str",",","Any","]",")",":","Specifies","the","runtime","environment","for","this","actor","or","task","and","its","children","(","see",":","ref",":","`","runtime-environments","`","for","details",")",".","max_pending_calls",":","Set","the","max","number","of","pending","calls","allowed","on","the","actor","handle",".","When","this","value","is","exceeded",",","PendingCallsLimitExceeded","will","be","raised","for","further","tasks",".","Note","that","this","limit","is","counted","per","handle",".","-1","means","that","the","number","of","pending","calls","is","unlimited",".","scheduling_strategy",":","Strategy","about","how","to","schedule","this","actor",".","Returns",":","A","handle","to","the","newly","created","actor","."]
code: def _remote(self, args=None, kwargs=None, **actor_options):
      """Create an actor.
  
      This method allows more flexibility than the remote method because
      resource requirements can be specified and override the defaults in the
      decorator.
  
      Args:
          args: The arguments to forward to the actor constructor.
          kwargs: The keyword arguments to forward to the actor constructor.
          num_cpus: The number of CPUs required by the actor creation task.
          num_gpus: The number of GPUs required by the actor creation task.
          memory: Restrict the heap memory usage of this actor.
          resources: The custom resources required by the actor creation
              task.
          max_concurrency: The max number of concurrent calls to allow for
              this actor. This only works with direct actor calls. The max
              concurrency defaults to 1 for threaded execution, and 1000 for
              asyncio execution. Note that the execution order is not
              guaranteed when max_concurrency > 1.
          name: The globally unique name for the actor, which can be used
              to retrieve the actor via ray.get_actor(name) as long as the
              actor is still alive.
          namespace: Override the namespace to use for the actor. By default,
              actors are created in an anonymous namespace. The actor can
              be retrieved via ray.get_actor(name=name, namespace=namespace).
          lifetime: Either `None`, which defaults to the actor will fate
              share with its creator and will be deleted once its refcount
              drops to zero, or "detached", which means the actor will live
              as a global object independent of the creator.
          placement_group: (This has been deprecated, please use
              `PlacementGroupSchedulingStrategy` scheduling_strategy)
              the placement group this actor belongs to,
              or None if it doesn't belong to any group. Setting to "default"
              autodetects the placement group based on the current setting of
              placement_group_capture_child_tasks.
          placement_group_bundle_index: (This has been deprecated, please use
              `PlacementGroupSchedulingStrategy` scheduling_strategy)
              the index of the bundle
              if the actor belongs to a placement group, which may be -1 to
              specify any available bundle.
          placement_group_capture_child_tasks: (This has been deprecated,
              please use `PlacementGroupSchedulingStrategy`
              scheduling_strategy)
              Whether or not children tasks
              of this actor should implicitly use the same placement group
              as its parent. It is False by default.
          runtime_env (Dict[str, Any]): Specifies the runtime environment for
              this actor or task and its children (see
              :ref:`runtime-environments` for details).
          max_pending_calls: Set the max number of pending calls
              allowed on the actor handle. When this value is exceeded,
              PendingCallsLimitExceeded will be raised for further tasks.
              Note that this limit is counted per handle. -1 means that the
              number of pending calls is unlimited.
          scheduling_strategy: Strategy about how to schedule this actor.
  
      Returns:
          A handle to the newly created actor.
      """
      name = actor_options.get("name")
      namespace = actor_options.get("namespace")
      if name is not None:
          if not isinstance(name, str):
              raise TypeError(
                  f"name must be None or a string, got: '{type(name)}'."
              )
          elif name == "":
              raise ValueError("Actor name cannot be an empty string.")
      if namespace is not None:
          ray._private.utils.validate_namespace(namespace)
  
      # Handle the get-or-create case.
      if actor_options.get("get_if_exists"):
          try:
              return ray.get_actor(name, namespace=namespace)
          except ValueError:
              # Attempt to create it (may race with other attempts).
              updated_options = actor_options.copy()
              updated_options[
                  "get_if_exists"
              ] = False  # prevent infinite loop
              try:
                  return self._remote(args, kwargs, **updated_options)
              except ValueError:
                  # We lost the creation race, ignore.
                  pass
              return ray.get_actor(name, namespace=namespace)
  
      # We pop the "concurrency_groups" coming from "@ray.remote" here. We no longer
      # need it in "_remote()".
      actor_options.pop("concurrency_groups", None)
  
      if args is None:
          args = []
      if kwargs is None:
          kwargs = {}
      meta = self.__ray_metadata__
      actor_has_async_methods = (
          len(
              inspect.getmembers(
                  meta.modified_class, predicate=is_async_func
              )
          )
          > 0
      )
      is_asyncio = actor_has_async_methods
  
      if actor_options.get("max_concurrency") is None:
          actor_options["max_concurrency"] = 1000 if is_asyncio else 1
  
      if client_mode_should_convert():
          return client_mode_convert_actor(
              self, args, kwargs, **actor_options
          )
  
      # fill actor required options
      for k, v in ray_option_utils.actor_options.items():
          actor_options[k] = actor_options.get(k, v.default_value)
      # "concurrency_groups" already takes effects and should not apply again.
      # Remove the default value here.
      actor_options.pop("concurrency_groups", None)
  
      # TODO(suquark): cleanup these fields
      max_concurrency = actor_options["max_concurrency"]
      lifetime = actor_options["lifetime"]
      runtime_env = actor_options["runtime_env"]
      placement_group = actor_options["placement_group"]
      placement_group_bundle_index = actor_options[
          "placement_group_bundle_index"
      ]
      placement_group_capture_child_tasks = actor_options[
          "placement_group_capture_child_tasks"
      ]
      scheduling_strategy = actor_options["scheduling_strategy"]
      max_restarts = actor_options["max_restarts"]
      max_task_retries = actor_options["max_task_retries"]
      max_pending_calls = actor_options["max_pending_calls"]
  
      if scheduling_strategy is None or not isinstance(
          scheduling_strategy, PlacementGroupSchedulingStrategy
      ):
          _warn_if_using_deprecated_placement_group(actor_options, 3)
  
      worker = ray._private.worker.global_worker
      worker.check_connected()
  
      # Check whether the name is already taken.
      # TODO(edoakes): this check has a race condition because two drivers
      # could pass the check and then create the same named actor. We should
      # instead check this when we create the actor, but that's currently an
      # async call.
      if name is not None:
          try:
              ray.get_actor(name, namespace=namespace)
          except ValueError:  # Name is not taken.
              pass
          else:
              raise ValueError(
                  f"The name {name} (namespace={namespace}) is already "
                  "taken. Please use "
                  "a different name or get the existing actor using "
                  f"ray.get_actor('{name}', namespace='{namespace}')"
              )
  
      if lifetime is None:
          detached = None
      elif lifetime == "detached":
          detached = True
      elif lifetime == "non_detached":
          detached = False
      else:
          raise ValueError(
              "actor `lifetime` argument must be one of 'detached', "
              "'non_detached' and 'None'."
          )
  
      # LOCAL_MODE cannot handle cross_language
      if worker.mode == ray.LOCAL_MODE:
          assert (
              not meta.is_cross_language
          ), "Cross language ActorClass cannot be executed locally."
  
      # Export the actor.
      if not meta.is_cross_language and (
          meta.last_export_session_and_job
          != worker.current_session_and_job
      ):
          # If this actor class was not exported in this session and job,
          # we need to export this function again, because current GCS
          # doesn't have it.
          meta.last_export_session_and_job = (
              worker.current_session_and_job
          )
          # After serialize / deserialize modified class, the __module__
          # of modified class will be ray.cloudpickle.cloudpickle.
          # So, here pass actor_creation_function_descriptor to make
          # sure export actor class correct.
          worker.function_actor_manager.export_actor_class(
              meta.modified_class,
              meta.actor_creation_function_descriptor,
              meta.method_meta.methods.keys(),
          )
  
      resources = ray._private.utils.resources_from_ray_options(
          actor_options
      )
      # Set the actor's default resources if not already set. First three
      # conditions are to check that no resources were specified in the
      # decorator. Last three conditions are to check that no resources were
      # specified when _remote() was called.
      # TODO(suquark): In the original code, memory is not considered as resources,
      # when deciding the default CPUs. It is strange, but we keep the original
      # semantics in case that it breaks user applications & tests.
      if not set(resources.keys()).difference(
          {"memory", "object_store_memory"}
      ):
          # In the default case, actors acquire no resources for
          # their lifetime, and actor methods will require 1 CPU.
          resources.setdefault(
              "CPU", ray_constants.DEFAULT_ACTOR_CREATION_CPU_SIMPLE
          )
          actor_method_cpu = (
              ray_constants.DEFAULT_ACTOR_METHOD_CPU_SIMPLE
          )
      else:
          # If any resources are specified (here or in decorator), then
          # all resources are acquired for the actor's lifetime and no
          # resources are associated with methods.
          resources.setdefault(
              "CPU", ray_constants.DEFAULT_ACTOR_CREATION_CPU_SPECIFIED
          )
          actor_method_cpu = (
              ray_constants.DEFAULT_ACTOR_METHOD_CPU_SPECIFIED
          )
  
      # If the actor methods require CPU resources, then set the required
      # placement resources. If actor_placement_resources is empty, then
      # the required placement resources will be the same as resources.
      actor_placement_resources = {}
      assert actor_method_cpu in [0, 1]
      if actor_method_cpu == 1:
          actor_placement_resources = resources.copy()
          actor_placement_resources["CPU"] += 1
      if meta.is_cross_language:
          creation_args = cross_language._format_args(
              worker, args, kwargs
          )
      else:
          function_signature = meta.method_meta.signatures["__init__"]
          creation_args = signature.flatten_args(
              function_signature, args, kwargs
          )
  
      if scheduling_strategy is None or isinstance(
          scheduling_strategy, PlacementGroupSchedulingStrategy
      ):
          # TODO(jjyao) Clean this up once the
          # placement_group option is removed.
          # We should also consider pushing this logic down to c++
          # so that it can be reused by all languages.
          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,
              actor_placement_resources,
              meta.class_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,
          )
  
      concurrency_groups_dict = {}
      if meta.concurrency_groups is None:
          meta.concurrency_groups = []
      for cg_name in meta.concurrency_groups:
          concurrency_groups_dict[cg_name] = {
              "name": cg_name,
              "max_concurrency": meta.concurrency_groups[cg_name],
              "function_descriptors": [],
          }
  
      # Update methods
      for method_name in meta.method_meta.concurrency_group_for_methods:
          cg_name = meta.method_meta.concurrency_group_for_methods[
              method_name
          ]
          assert cg_name in concurrency_groups_dict
  
          module_name = (
              meta.actor_creation_function_descriptor.module_name
          )
          class_name = (
              meta.actor_creation_function_descriptor.class_name
          )
          concurrency_groups_dict[cg_name][
              "function_descriptors"
          ].append(
              PythonFunctionDescriptor(
                  module_name, method_name, class_name
              )
          )
  
      # Update the creation descriptor based on number of arguments
      if meta.is_cross_language:
          func_name = "<init>"
          if meta.language == Language.CPP:
              func_name = (
                  meta.actor_creation_function_descriptor.function_name
              )
          meta.actor_creation_function_descriptor = (
              cross_language._get_function_descriptor_for_actor_method(
                  meta.language,
                  meta.actor_creation_function_descriptor,
                  func_name,
                  str(len(args) + len(kwargs)),
              )
          )
  
      actor_id = worker.core_worker.create_actor(
          meta.language,
          meta.actor_creation_function_descriptor,
          creation_args,
          max_restarts,
          max_task_retries,
          resources,
          actor_placement_resources,
          max_concurrency,
          detached,
          name if name is not None else "",
          namespace if namespace is not None else "",
          is_asyncio,
          # Store actor_method_cpu in actor handle's extension data.
          extension_data=str(actor_method_cpu),
          serialized_runtime_env_info=serialized_runtime_env_info
          or "{}",
          concurrency_groups_dict=concurrency_groups_dict or dict(),
          max_pending_calls=max_pending_calls,
          scheduling_strategy=scheduling_strategy,
      )
  
      if _actor_launch_hook:
          _actor_launch_hook(
              meta.actor_creation_function_descriptor,
              resources,
              scheduling_strategy,
          )
  
      actor_handle = ActorHandle(
          meta.language,
          actor_id,
          max_task_retries,
          meta.method_meta.method_is_generator,
          meta.method_meta.decorators,
          meta.method_meta.signatures,
          meta.method_meta.num_returns,
          meta.method_meta.max_retries,
          meta.method_meta.retry_exceptions,
          meta.method_meta.generator_backpressure_num_objects,
          actor_method_cpu,
          meta.actor_creation_function_descriptor,
          worker.current_session_and_job,
          original_handle=True,
      )
  
      return actor_handle
  
imports: import inspect
  import logging
  import weakref
  from typing import Any, Dict, List, Optional, Union
  import ray._private.ray_constants
  import ray._private.signature
  import ray._private.worker
  import ray._raylet
  from ray import ActorClassID, Language, cross_language
  from ray._private import ray_option_utils
  from ray._private.async_compat import is_async_func
  from ray._private.auto_init_hook import wrap_auto_init
  from ray._private.client_mode_hook import client_mode_convert_actor, client_mode_hook, client_mode_should_convert
  from ray._private.inspect_util import is_class_method, is_function_or_method, is_static_method
  from ray._private.ray_option_utils import _warn_if_using_deprecated_placement_group
  from ray._private.utils import get_runtime_env_info, parse_runtime_env
  from ray._raylet import STREAMING_GENERATOR_RETURN, ObjectRefGenerator, PythonFunctionDescriptor, raise_sys_exit_with_custom_error_message
  from ray.exceptions import AsyncioActorExit
  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, SchedulingStrategyT
  from ray.util.tracing.tracing_helper import _inject_tracing_into_class, _tracing_actor_creation, _tracing_actor_method_invocation
label_desc: Use image node_id 7 for calling the ActorClass obj's underlying member method code with example usage: obj._remote(args, kwargs) and returns: actor_handle, client_mode_convert_actor, ray, ray, self