python/ray/actor.py
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.
            object_store_memory: Restrict the object store memory used by
                this actor when creating objects.
            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 (int): 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.
        """
        # 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=inspect.iscoroutinefunction
                )
            )
            > 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(auto_init=True):
            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"]
        name = actor_options["name"]
        namespace = actor_options["namespace"]
        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"]

        worker = ray.worker.global_worker
        worker.check_connected()

        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)

        # 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,
        )

        actor_handle = ActorHandle(
            meta.language,
            actor_id,
            meta.method_meta.decorators,
            meta.method_meta.signatures,
            meta.method_meta.num_returns,
            actor_method_cpu,
            meta.actor_creation_function_descriptor,
            worker.current_session_and_job,
            original_handle=True,
        )

        return actor_handle
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.
            object_store_memory: Restrict the object store memory used by
                this actor when creating objects.
            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 (int): 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.
        
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.
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 .
def _remote(self, args=None, kwargs=None, **actor_options):
        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=inspect.iscoroutinefunction
                )
            )
            > 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(auto_init=True):
            return client_mode_convert_actor(self, args, kwargs, **actor_options)
        for k, v in ray_option_utils.actor_options.items():
            actor_options[k] = actor_options.get(k, v.default_value)
        actor_options.pop("concurrency_groups", None)
        max_concurrency = actor_options["max_concurrency"]
        name = actor_options["name"]
        namespace = actor_options["namespace"]
        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"]
        worker = ray.worker.global_worker
        worker.check_connected()
        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)
        if name is not None:
            try:
                ray.get_actor(name, namespace=namespace)
            except ValueError:  
                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'."
            )
        if worker.mode == ray.LOCAL_MODE:
            assert (
                not meta.is_cross_language
            ), "Cross language ActorClass cannot be executed locally."
        if not meta.is_cross_language and (
            meta.last_export_session_and_job != worker.current_session_and_job
        ):
            meta.last_export_session_and_job = worker.current_session_and_job
            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)
        if not set(resources.keys()).difference({"memory", "object_store_memory"}):
            resources.setdefault("CPU", ray_constants.DEFAULT_ACTOR_CREATION_CPU_SIMPLE)
            actor_method_cpu = ray_constants.DEFAULT_ACTOR_METHOD_CPU_SIMPLE
        else:
            resources.setdefault(
                "CPU", ray_constants.DEFAULT_ACTOR_CREATION_CPU_SPECIFIED
            )
            actor_method_cpu = ray_constants.DEFAULT_ACTOR_METHOD_CPU_SPECIFIED
        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
        ):
            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": [],
            }
        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)
            )
        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,
            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,
        )
        actor_handle = ActorHandle(
            meta.language,
            actor_id,
            meta.method_meta.decorators,
            meta.method_meta.signatures,
            meta.method_meta.num_returns,
            actor_method_cpu,
            meta.actor_creation_function_descriptor,
            worker.current_session_and_job,
            original_handle=True,
        )
        return actor_handle
def _remote ( self , args = None , kwargs = None , ** actor_options ) : 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 = inspect . iscoroutinefunction ) ) > 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 ( auto_init = True ) : return client_mode_convert_actor ( self , args , kwargs , ** actor_options ) for k , v in ray_option_utils . actor_options . items ( ) : actor_options [ k ] = actor_options . get ( k , v . default_value ) actor_options . pop ( "concurrency_groups" , None ) max_concurrency = actor_options [ "max_concurrency" ] name = actor_options [ "name" ] namespace = actor_options [ "namespace" ] 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" ] worker = ray . worker . global_worker worker . check_connected ( ) 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 ) if name is not None : try : ray . get_actor ( name , namespace = namespace ) except ValueError : 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'." ) if worker . mode == ray . LOCAL_MODE : assert ( not meta . is_cross_language ) , "Cross language ActorClass cannot be executed locally." if not meta . is_cross_language and ( meta . last_export_session_and_job != worker . current_session_and_job ) : meta . last_export_session_and_job = worker . current_session_and_job 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 ) if not set ( resources . keys ( ) ) . difference ( { "memory" , "object_store_memory" } ) : resources . setdefault ( "CPU" , ray_constants . DEFAULT_ACTOR_CREATION_CPU_SIMPLE ) actor_method_cpu = ray_constants . DEFAULT_ACTOR_METHOD_CPU_SIMPLE else : resources . setdefault ( "CPU" , ray_constants . DEFAULT_ACTOR_CREATION_CPU_SPECIFIED ) actor_method_cpu = ray_constants . DEFAULT_ACTOR_METHOD_CPU_SPECIFIED 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 ) : 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" : [ ] , } 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 ) ) 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 , 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 , ) actor_handle = ActorHandle ( meta . language , actor_id , meta . method_meta . decorators , meta . method_meta . signatures , meta . method_meta . num_returns , actor_method_cpu , meta . actor_creation_function_descriptor , worker . current_session_and_job , original_handle = True , ) return actor_handle
Create an actor.
Create an actor .
"""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.
            object_store_memory: Restrict the object store memory used by
                this actor when creating objects.
            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 (int): 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.
        """ # We pop the "concurrency_groups" coming from "@ray.remote" here. We no longer # need it in "_remote()". # fill actor required options # "concurrency_groups" already takes effects and should not apply again. # Remove the default value here. # TODO(suquark): cleanup these fields # 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. # Name is not taken. # LOCAL_MODE cannot handle cross_language # Export the actor. # 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. # 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. # 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. # In the default case, actors acquire no resources for # their lifetime, and actor methods will require 1 CPU. # 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. # 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. # 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. # Update methods # Update the creation descriptor based on number of arguments # Store actor_method_cpu in actor handle's extension data.
A handle to the newly created actor.
A handle to the newly created actor .
self
args
The arguments to forward to the actor constructor.
The arguments to forward to the actor constructor .
kwargs
The keyword arguments to forward to the actor constructor.
The keyword arguments to forward to the actor constructor .
num_cpus
The number of CPUs required by the actor creation task.
The number of CPUs required by the actor creation task .
num_gpus
The number of GPUs required by the actor creation task.
The number of GPUs required by the actor creation task .
memory
Restrict the heap memory usage of this actor.
Restrict the heap memory usage of this actor .
object_store_memory
Restrict the object store memory used by
this actor when creating objects.
Restrict the object store memory used by this actor when creating objects .
resources
The custom resources required by the actor creation
task.
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.
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.
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).
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.
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.
( 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.
( 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.
( 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
Specifies the runtime environment for
this actor or task and its children (see
:ref:`runtime-environments` for details).
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.
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.
Strategy about how to schedule this actor .