class_docstr: Management class for Ray actors and actor tasks.
  
  This class provides an event-based management interface for actors, and
  actor tasks.
  
  The manager can be used to start actors, stop actors, and schedule and
  track task futures on these actors.
  The manager will then invoke callbacks related to the tracked entities.
  
  For instance, when an actor is added with
  :meth:`add_actor() <RayActorManager.add_actor>`,
  a :ref:`TrackedActor <ray.air.execution._internal.tracked_actor.TrackedActor`
  object is returned. An ``on_start`` callback can be specified that is invoked
  once the actor successfully started. Similarly, ``on_stop`` and ``on_error``
  can be used to specify callbacks relating to the graceful or ungraceful
  end of an actor's lifetime.
  
  When scheduling an actor task using
  :meth:`schedule_actor_task()
  <ray.air.execution._internal.actor_manager.RayActorManager.schedule_actor_task>`,
  an ``on_result`` callback can be specified that is invoked when the task
  successfully resolves, and an ``on_error`` callback will resolve when the
  task fails.
  
  The RayActorManager does not implement any true asynchronous processing. Control
  has to be explicitly yielded to the event manager via :meth:`RayActorManager.next`.
  Callbacks will only be invoked when control is with the RayActorManager, and
  callbacks will always be executed sequentially in order of arriving events.
  
  Args:
      resource_manager: Resource manager used to request resources for the actors.
  
  Example:
  
      .. code-block:: python
  
          from ray.air.execution import ResourceRequest
          from ray.air.execution._internal import RayActorManager
  
          actor_manager = RayActorManager()
  
          # Request an actor
          tracked_actor = actor_manager.add_actor(
              ActorClass,
              kwargs={},
              resource_request=ResourceRequest([{"CPU": 1}]),
              on_start=actor_start_callback,
              on_stop=actor_stop_callback,
              on_error=actor_error_callback
          )
  
          # Yield control to event manager to start actor
          actor_manager.next()
  
          # Start task on the actor (ActorClass.foo.remote())
          tracked_actor_task = actor_manager.schedule_actor_task(
              tracked_actor,
              method_name="foo",
              on_result=task_result_callback,
              on_error=task_error_callback
          )
  
          # Again yield control to event manager to process task futures
          actor_manager.wait()
class_docstr_tok: ["Management","class","for","Ray","actors","and","actor","tasks",".","This","class","provides","an","event-based","management","interface","for","actors",",","and","actor","tasks",".","The","manager","can","be","used","to","start","actors",",","stop","actors",",","and","schedule","and","track","task","futures","on","these","actors",".","The","manager","will","then","invoke","callbacks","related","to","the","tracked","entities",".","For","instance",",","when","an","actor","is","added","with",":","meth",":","`","add_actor","(",")","<","RayActorManager.add_actor",">","`",",","a",":","ref",":","`","TrackedActor","<","ray.air.execution._internal.tracked_actor.TrackedActor","`","object","is","returned",".","An","``","on_start","``","callback","can","be","specified","that","is","invoked","once","the","actor","successfully","started",".","Similarly",",","``","on_stop","``","and","``","on_error","``","can","be","used","to","specify","callbacks","relating","to","the","graceful","or","ungraceful","end","of","an","actor","'s","lifetime",".","When","scheduling","an","actor","task","using",":","meth",":","`","schedule_actor_task","(",")","<","ray.air.execution._internal.actor_manager.RayActorManager.schedule_actor_task",">","`",",","an","``","on_result","``","callback","can","be","specified","that","is","invoked","when","the","task","successfully","resolves",",","and","an","``","on_error","``","callback","will","resolve","when","the","task","fails",".","The","RayActorManager","does","not","implement","any","true","asynchronous","processing",".","Control","has","to","be","explicitly","yielded","to","the","event","manager","via",":","meth",":","`","RayActorManager.next","`",".","Callbacks","will","only","be","invoked","when","control","is","with","the","RayActorManager",",","and","callbacks","will","always","be","executed","sequentially","in","order","of","arriving","events",".","Args",":","resource_manager",":","Resource","manager","used","to","request","resources","for","the","actors",".","Example",":","..","code-block",":",":","python","from","ray.air.execution","import","ResourceRequest","from","ray.air.execution._internal","import","RayActorManager","actor_manager","=","RayActorManager","(",")","#","Request","an","actor","tracked_actor","=","actor_manager.add_actor","(","ActorClass",",","kwargs=","{","}",",","resource_request=ResourceRequest","(","[","{","``","CPU","''",":","1","}","]",")",",","on_start=actor_start_callback",",","on_stop=actor_stop_callback",",","on_error=actor_error_callback",")","#","Yield","control","to","event","manager","to","start","actor","actor_manager.next","(",")","#","Start","task","on","the","actor","(","ActorClass.foo.remote","(",")",")","tracked_actor_task","=","actor_manager.schedule_actor_task","(","tracked_actor",",","method_name=","''","foo","''",",","on_result=task_result_callback",",","on_error=task_error_callback",")","#","Again","yield","control","to","event","manager","to","process","task","futures","actor_manager.wait","(",")"]
code: def __init__(self, resource_manager: ResourceManager):
      self._resource_manager: ResourceManager = resource_manager
  
      self._actor_state_events = RayEventManager()
      self._actor_task_events = RayEventManager()
  
      # ---
      # Tracked actor futures.
  
      # This maps TrackedActor objects to their futures. We use this to see if an
      # actor has any futures scheduled and to remove them when we terminate an actor.
  
      # Actors to actor task futures
      self._tracked_actors_to_task_futures: Dict[
          TrackedActor, Set[ray.ObjectRef]
      ] = defaultdict(set)
  
      # Actors to actor state futures (start/terminate)
      self._tracked_actors_to_state_futures: Dict[
          TrackedActor, Set[ray.ObjectRef]
      ] = defaultdict(set)
  
      # ---
      # Pending actors.
      # We use three dicts for actors that are requested but not yet started.
  
      # This dict keeps a list of actors associated with each resource request.
      # We use this to start actors in the correct order when their resources
      # become available.
      self._resource_request_to_pending_actors: Dict[
          ResourceRequest, List[TrackedActor]
      ] = defaultdict(list)
  
      # This dict stores the actor class, kwargs, and resource request of
      # pending actors. Once the resources are available, we start the remote
      # actor class with its args. We need the resource request to cancel it
      # if needed.
      self._pending_actors_to_attrs: Dict[
          TrackedActor, Tuple[Type, Dict[str, Any], ResourceRequest]
      ] = {}
  
      # This dict keeps track of cached actor tasks. We can't schedule actor
      # tasks before the actor is actually scheduled/live. So when the caller
      # tries to schedule a task, we cache it here, and schedule it once the
      # actor is started.
      self._pending_actors_to_enqueued_actor_tasks: Dict[
          TrackedActor,
          List[
              Tuple[TrackedActorTask, str, Tuple[Any], Dict[str, Any]]
          ],
      ] = defaultdict(list)
  
      # ---
      # Live actors.
      # We keep one dict for actors that are currently running and a set of
      # actors that we should forcefully kill.
  
      # This dict associates the TrackedActor object with the Ray actor handle
      # and the resources associated to the actor. We use it to schedule the
      # actual ray tasks, and to return the resources when the actor stopped.
      self._live_actors_to_ray_actors_resources: Dict[
          TrackedActor, Tuple[ray.actor.ActorHandle, AcquiredResources]
      ] = {}
      self._live_resource_cache: Optional[Dict[str, Any]] = None
  
      # This dict contains all actors that should be killed (after calling
      # `remove_actor()`). Kill requests will be handled in wait().
      self._live_actors_to_kill: Set[TrackedActor] = set()
  
      # Track failed actors
      self._failed_actor_ids: Set[int] = set()
  
imports: import logging
  import random
  import time
  import uuid
  from collections import defaultdict, Counter
  from typing import Any, Callable, Dict, List, Optional, Set, Tuple, Type, Union
  import ray
  from ray.air.execution._internal.event_manager import RayEventManager
  from ray.air.execution.resources import AcquiredResources, ResourceManager, ResourceRequest
  from ray.air.execution._internal.tracked_actor import TrackedActor
  from ray.air.execution._internal.tracked_actor_task import TrackedActorTask
  from ray.exceptions import RayTaskError, RayActorError
label_desc: Use image node_id 1 to create a new RayActorManager object  with example: obj = RayActorManager(resource_manager)