class_docstr: A manager that is aware of the healthiness of remote actors.
  
  .. testcode::
      :skipif: True
  
      import ray
      from ray.rllib.utils.actor_manager import FaultTolerantActorManager
  
      @ray.remote
      class MyActor:
      def apply(self, fn) -> Any:
          return fn(self)
  
      def do_something(self):
          return True
  
      actors = [MyActor.remote() for _ in range(3)]
      manager = FaultTolerantActorManager(
          actors, max_remote_requests_in_flight_per_actor=2,
      )
  
      # Synchronous remote calls.
      results = manager.foreach_actor(lambda actor: actor.do_something())
      # Print results ignoring returned errors.
      print([r.get() for r in results.ignore_errors()])
  
      # Asynchronous remote calls.
      manager.foreach_actor_async(lambda actor: actor.do_something())
      time.sleep(2) # Wait for the tasks to finish.
      for r in manager.fetch_ready_async_reqs()
          # Handle result and errors.
          if r.ok: print(r.get())
          else print("Error: {}".format(r.get()))
class_docstr_tok: ["A","manager","that","is","aware","of","the","healthiness","of","remote","actors",".","..","testcode",":",":",":","skipif",":","True","import","ray","from","ray.rllib.utils.actor_manager","import","FaultTolerantActorManager","@","ray.remote","class","MyActor",":","def","apply","(","self",",","fn",")","-",">","Any",":","return","fn","(","self",")","def","do_something","(","self",")",":","return","True","actors","=","[","MyActor.remote","(",")","for","_","in","range","(","3",")","]","manager","=","FaultTolerantActorManager","(","actors",",","max_remote_requests_in_flight_per_actor=2",",",")","#","Synchronous","remote","calls",".","results","=","manager.foreach_actor","(","lambda","actor",":","actor.do_something","(",")",")","#","Print","results","ignoring","returned","errors",".","print","(","[","r.get","(",")","for","r","in","results.ignore_errors","(",")","]",")","#","Asynchronous","remote","calls",".","manager.foreach_actor_async","(","lambda","actor",":","actor.do_something","(",")",")","time.sleep","(","2",")","#","Wait","for","the","tasks","to","finish",".","for","r","in","manager.fetch_ready_async_reqs","(",")","#","Handle","result","and","errors",".","if","r.ok",":","print","(","r.get","(",")",")","else","print","(","``","Error",":","{","}","''",".format","(","r.get","(",")",")",")"]
docstr: Construct a FaultTolerantActorManager.
  
  Args:
      actors: A list of ray remote actors to manage on. These actors must have an
          ``apply`` method which takes a function with only one parameter (the
          actor instance itself).
      max_remote_requests_in_flight_per_actor: The maximum number of remote
          requests that can be in flight per actor. Any requests made to the pool
          that cannot be scheduled because the limit has been reached will be
          dropped. This only applies to the asynchronous remote call mode.
      init_id: The initial ID to use for the next remote actor. Default is 0.
docstr_tok: ["Construct","a","FaultTolerantActorManager",".","Args",":","actors",":","A","list","of","ray","remote","actors","to","manage","on",".","These","actors","must","have","an","``","apply","``","method","which","takes","a","function","with","only","one","parameter","(","the","actor","instance","itself",")",".","max_remote_requests_in_flight_per_actor",":","The","maximum","number","of","remote","requests","that","can","be","in","flight","per","actor",".","Any","requests","made","to","the","pool","that","can","not","be","scheduled","because","the","limit","has","been","reached","will","be","dropped",".","This","only","applies","to","the","asynchronous","remote","call","mode",".","init_id",":","The","initial","ID","to","use","for","the","next","remote","actor",".","Default","is","0","."]
code: def __init__(
      self,
      actors: Optional[List[ActorHandle]] = None,
      max_remote_requests_in_flight_per_actor: int = 2,
      init_id: int = 0,
  ):
      """Construct a FaultTolerantActorManager.
  
      Args:
          actors: A list of ray remote actors to manage on. These actors must have an
              ``apply`` method which takes a function with only one parameter (the
              actor instance itself).
          max_remote_requests_in_flight_per_actor: The maximum number of remote
              requests that can be in flight per actor. Any requests made to the pool
              that cannot be scheduled because the limit has been reached will be
              dropped. This only applies to the asynchronous remote call mode.
          init_id: The initial ID to use for the next remote actor. Default is 0.
      """
      # For historic reasons, just start remote worker ID from 1, so they never
      # collide with local worker ID (0).
      self.__next_id = init_id
  
      # Actors are stored in a map and indexed by a unique id.
      self.__actors: Mapping[int, ActorHandle] = {}
      self.__remote_actor_states: Mapping[int, self._ActorState] = {}
      self.__restored_actors = set()
      self.add_actors(actors or [])
  
      # Maps outstanding async requests to the ids of the actors that
      # are executing them.
      self.__in_flight_req_to_actor_id: Mapping[ray.ObjectRef, int] = {}
  
      self._max_remote_requests_in_flight_per_actor = (
          max_remote_requests_in_flight_per_actor
      )
  
      # Useful metric.
      self._num_actor_restarts = 0
  
imports: from collections import defaultdict
  import copy
  from dataclasses import dataclass
  import logging
  import sys
  import time
  from typing import Any, Callable, Dict, Iterator, List, Mapping, Optional, Tuple, Union
  import ray
  from ray.actor import ActorHandle
  from ray.exceptions import RayActorError, RayError, RayTaskError
  from ray.rllib.utils.typing import T
  from ray.util.annotations import DeveloperAPI
label_desc: Use image node_id 1 to create a new FaultTolerantActorManager object  with example: obj = FaultTolerantActorManager(actors, max_remote_requests_in_flight_per_actor, init_id)