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: Yield control to event manager to await the next event and invoke callbacks.
  
  Calling this method will wait for up to ``timeout`` seconds for the next
  event to arrive.
  
  When events arrive, callbacks relating to the events will be
  invoked. A timeout of ``None`` will block until the next event arrives.
  
  Note:
      If an actor task fails with a ``RayActorError``, this is one event,
      but it may trigger _two_ `on_error` callbacks: One for the actor,
      and one for the task.
  
  Note:
      The ``timeout`` argument is used for pure waiting time for events. It does
      not include time spent on processing callbacks. Depending on the processing
      time of the callbacks, it can take much longer for this function to
      return than the specified timeout.
  
  Args:
      timeout: Timeout in seconds to wait for next event.
  
  Returns:
      True if at least one event was processed.
code: def next(self, timeout: Optional[Union[int, float]] = None) -> bool:
      """Yield control to event manager to await the next event and invoke callbacks.
  
      Calling this method will wait for up to ``timeout`` seconds for the next
      event to arrive.
  
      When events arrive, callbacks relating to the events will be
      invoked. A timeout of ``None`` will block until the next event arrives.
  
      Note:
          If an actor task fails with a ``RayActorError``, this is one event,
          but it may trigger _two_ `on_error` callbacks: One for the actor,
          and one for the task.
  
      Note:
          The ``timeout`` argument is used for pure waiting time for events. It does
          not include time spent on processing callbacks. Depending on the processing
          time of the callbacks, it can take much longer for this function to
          return than the specified timeout.
  
      Args:
          timeout: Timeout in seconds to wait for next event.
  
      Returns:
          True if at least one event was processed.
  
      """
      # First issue any pending forceful actor kills
      actor_killed = self._try_kill_actor()
  
      # We always try to start actors as this won't trigger an event callback
      self._try_start_actors()
  
      # If an actor was killed, this was our event, and we return.
      if actor_killed:
          return True
  
      # Otherwise, collect all futures and await the next.
      resource_futures = self._resource_manager.get_resource_futures()
      actor_state_futures = self._actor_state_events.get_futures()
      actor_task_futures = self._actor_task_events.get_futures()
  
      # Shuffle state futures
      shuffled_state_futures = list(actor_state_futures)
      random.shuffle(shuffled_state_futures)
  
      # Shuffle task futures
      shuffled_task_futures = list(actor_task_futures)
      random.shuffle(shuffled_task_futures)
  
      # Prioritize resource futures over actor state over task futures
      all_futures = (
          resource_futures
          + shuffled_state_futures
          + shuffled_task_futures
      )
  
      start_wait = time.monotonic()
      ready, _ = ray.wait(all_futures, num_returns=1, timeout=timeout)
  
      if not ready:
          return False
  
      [future] = ready
  
      if future in actor_state_futures:
          self._actor_state_events.resolve_future(future)
      elif future in actor_task_futures:
          self._actor_task_events.resolve_future(future)
      else:
          self._handle_ready_resource_future()
          # Ready resource futures don't count as one event as they don't trigger
          # any callbacks. So we repeat until we hit anything that is not a resource
          # future.
          time_taken = time.monotonic() - start_wait
          return self.next(
              timeout=max(1e-9, timeout - time_taken)
              if timeout is not None
              else None
          )
  
      self._try_start_actors()
      return True
  
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 2 for calling the RayActorManager obj's underlying member method code with example usage: obj.next(timeout) and returns: True, True, False, self