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","(",")"]
docstr: Schedule and track a task on an actor.
  
  This method will schedule a remote task ``method_name`` on the
  ``tracked_actor``.
  
  This method accepts two optional callbacks that will be invoked when
  their respective events are triggered.
  
  The ``on_result`` callback is triggered when a task resolves successfully.
  It should accept two arguments: The actor for which the
  task resolved, and the result received from the remote call.
  
  The ``on_error`` callback is triggered when a task fails.
  It should accept two arguments: The actor for which the
  task threw an error, and the exception.
  
  Args:
      tracked_actor: Actor to schedule task on.
      method_name: Remote method name to invoke on the actor. If this is
          e.g. ``foo``, then ``actor.foo.remote(*args, **kwargs)`` will be
          scheduled.
      args: Arguments to pass to the task.
      kwargs: Keyword arguments to pass to the task.
      on_result: Callback to invoke when the task resolves.
      on_error: Callback to invoke when the task fails.
  
  Raises:
      ValueError: If the ``tracked_actor`` is not managed by this event manager.
code: def schedule_actor_task(
      self,
      tracked_actor: TrackedActor,
      method_name: str,
      args: Optional[Tuple] = None,
      kwargs: Optional[Dict] = None,
      on_result: Optional[Callable[[TrackedActor, Any], None]] = None,
      on_error: Optional[
          Callable[[TrackedActor, Exception], None]
      ] = None,
      _return_future: bool = False,
  ) -> Optional[ray.ObjectRef]:
      """Schedule and track a task on an actor.
  
      This method will schedule a remote task ``method_name`` on the
      ``tracked_actor``.
  
      This method accepts two optional callbacks that will be invoked when
      their respective events are triggered.
  
      The ``on_result`` callback is triggered when a task resolves successfully.
      It should accept two arguments: The actor for which the
      task resolved, and the result received from the remote call.
  
      The ``on_error`` callback is triggered when a task fails.
      It should accept two arguments: The actor for which the
      task threw an error, and the exception.
  
      Args:
          tracked_actor: Actor to schedule task on.
          method_name: Remote method name to invoke on the actor. If this is
              e.g. ``foo``, then ``actor.foo.remote(*args, **kwargs)`` will be
              scheduled.
          args: Arguments to pass to the task.
          kwargs: Keyword arguments to pass to the task.
          on_result: Callback to invoke when the task resolves.
          on_error: Callback to invoke when the task fails.
  
      Raises:
          ValueError: If the ``tracked_actor`` is not managed by this event manager.
  
      """
      args = args or tuple()
      kwargs = kwargs or {}
  
      if tracked_actor.actor_id in self._failed_actor_ids:
          return
  
      tracked_actor_task = TrackedActorTask(
          tracked_actor=tracked_actor,
          on_result=on_result,
          on_error=on_error,
      )
  
      if tracked_actor not in self._live_actors_to_ray_actors_resources:
          # Actor is not started, yet
          if tracked_actor not in self._pending_actors_to_attrs:
              raise ValueError(
                  f"Tracked actor is not managed by this event manager: "
                  f"{tracked_actor}"
              )
  
          # Cache tasks for future execution
          self._pending_actors_to_enqueued_actor_tasks[
              tracked_actor
          ].append((tracked_actor_task, method_name, args, kwargs))
      else:
          res = self._schedule_tracked_actor_task(
              tracked_actor_task=tracked_actor_task,
              method_name=method_name,
              args=args,
              kwargs=kwargs,
              _return_future=_return_future,
          )
          if _return_future:
              return res[1]
  
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 26 for calling the RayActorManager obj's underlying member method code with example usage: obj.schedule_actor_task(tracked_actor, method_name, args, kwargs, on_result, on_error, _return_future) and returns: res