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: Remove a tracked actor.
  
  If the actor has already been started, this will stop the actor. This will
  trigger the :meth:`TrackedActor.on_stop
  <ray.air.execution._internal.tracked_actor.TrackedActor.on_stop>`
  callback once the actor stopped.
  
  If the actor has only been requested, but not started, yet, this will cancel
  the actor request. This will not trigger any callback.
  
  If ``kill=True``, this will use ``ray.kill()`` to forcefully terminate the
  actor. Otherwise, graceful actor deconstruction will be scheduled after
  all currently tracked futures are resolved.
  
  This method returns a boolean, indicating if a stop future is tracked and
  the ``on_stop`` callback will be invoked. If the actor has been alive,
  this will be ``True``. If the actor hasn't been scheduled, yet, or failed
  (and triggered the ``on_error`` callback), this will be ``False``.
  
  Args:
      tracked_actor: Tracked actor to be removed.
      kill: If set, will forcefully terminate the actor instead of gracefully
          scheduling termination.
      stop_future: If set, use this future to track actor termination.
          Otherwise, schedule a ``__ray_terminate__`` future.
  
  Returns:
      Boolean indicating if the actor was previously alive, and thus whether
      a callback will be invoked once it is terminated.
code: def remove_actor(
      self,
      tracked_actor: TrackedActor,
      kill: bool = False,
      stop_future: Optional[ray.ObjectRef] = None,
  ) -> bool:
      """Remove a tracked actor.
  
      If the actor has already been started, this will stop the actor. This will
      trigger the :meth:`TrackedActor.on_stop
      <ray.air.execution._internal.tracked_actor.TrackedActor.on_stop>`
      callback once the actor stopped.
  
      If the actor has only been requested, but not started, yet, this will cancel
      the actor request. This will not trigger any callback.
  
      If ``kill=True``, this will use ``ray.kill()`` to forcefully terminate the
      actor. Otherwise, graceful actor deconstruction will be scheduled after
      all currently tracked futures are resolved.
  
      This method returns a boolean, indicating if a stop future is tracked and
      the ``on_stop`` callback will be invoked. If the actor has been alive,
      this will be ``True``. If the actor hasn't been scheduled, yet, or failed
      (and triggered the ``on_error`` callback), this will be ``False``.
  
      Args:
          tracked_actor: Tracked actor to be removed.
          kill: If set, will forcefully terminate the actor instead of gracefully
              scheduling termination.
          stop_future: If set, use this future to track actor termination.
              Otherwise, schedule a ``__ray_terminate__`` future.
  
      Returns:
          Boolean indicating if the actor was previously alive, and thus whether
          a callback will be invoked once it is terminated.
  
      """
      if tracked_actor.actor_id in self._failed_actor_ids:
          logger.debug(
              f"Tracked actor already failed, no need to remove: {tracked_actor}"
          )
          return False
      elif tracked_actor in self._live_actors_to_ray_actors_resources:
          # Ray actor is running.
  
          if not kill:
              # Schedule __ray_terminate__ future
              ray_actor, _ = self._live_actors_to_ray_actors_resources[
                  tracked_actor
              ]
  
              # Clear state futures here to avoid resolving __ray_ready__ futures
              for future in list(
                  self._tracked_actors_to_state_futures[tracked_actor]
              ):
                  self._actor_state_events.discard_future(future)
                  self._tracked_actors_to_state_futures[
                      tracked_actor
                  ].remove(future)
  
                  # If the __ray_ready__ future hasn't resolved yet, but we already
                  # scheduled the actor via Actor.remote(), we just want to stop
                  # it but not trigger any callbacks. This is in accordance with
                  # the contract defined in the docstring.
                  tracked_actor._on_start = None
                  tracked_actor._on_stop = None
                  tracked_actor._on_error = None
  
              def on_actor_stop(*args, **kwargs):
                  self._actor_stop_resolved(tracked_actor=tracked_actor)
  
              if stop_future:
                  # If the stop future was schedule via the actor manager,
                  # discard (track it as state future instead).
                  self._actor_task_events.discard_future(stop_future)
              else:
                  stop_future = ray_actor.__ray_terminate__.remote()
  
              self._actor_state_events.track_future(
                  future=stop_future,
                  on_result=on_actor_stop,
                  on_error=on_actor_stop,
              )
  
              self._tracked_actors_to_state_futures[tracked_actor].add(
                  stop_future
              )
          else:
              # kill = True
              self._live_actors_to_kill.add(tracked_actor)
  
          return True
  
      elif tracked_actor in self._pending_actors_to_attrs:
          # Actor is pending, stop
          _, _, resource_request = self._pending_actors_to_attrs.pop(
              tracked_actor
          )
          self._resource_request_to_pending_actors[
              resource_request
          ].remove(tracked_actor)
          self._resource_manager.cancel_resource_request(
              resource_request=resource_request
          )
          return False
      else:
          raise ValueError(f"Unknown tracked actor: {tracked_actor}")
  
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 22 for calling the RayActorManager obj's underlying member method code with example usage: obj.remove_actor(tracked_actor, kill, stop_future) and returns: False, True, False