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: Get results from outstanding async requests that are ready.
  
  Automatically mark actors unhealthy if they fail to respond.
  
  Note: If tags is an empty tuple then results from all ready async requests are
  returned.
  
  Args:
      timeout_seconds: Ray.get() timeout. Default is 0 (only those that are
          already ready).
      tags: A tag or a list of tags to identify the results from this async call.
      return_obj_refs: Whether to return ObjectRef instead of actual results.
      mark_healthy: whether to mark certain actors healthy based on the results
          of these remote calls. Useful, for example, to make sure actors
          do not come back without proper state restoration.
  
  Returns:
      A list of return values of all calls to `func(actor)` that are ready.
      The values may be actual data returned or exceptions raised during the
      remote call in the format of RemoteCallResults.
docstr_tok: ["Get","results","from","outstanding","async","requests","that","are","ready",".","Automatically","mark","actors","unhealthy","if","they","fail","to","respond",".","Note",":","If","tags","is","an","empty","tuple","then","results","from","all","ready","async","requests","are","returned",".","Args",":","timeout_seconds",":","Ray.get","(",")","timeout",".","Default","is","0","(","only","those","that","are","already","ready",")",".","tags",":","A","tag","or","a","list","of","tags","to","identify","the","results","from","this","async","call",".","return_obj_refs",":","Whether","to","return","ObjectRef","instead","of","actual","results",".","mark_healthy",":","whether","to","mark","certain","actors","healthy","based","on","the","results","of","these","remote","calls",".","Useful",",","for","example",",","to","make","sure","actors","do","not","come","back","without","proper","state","restoration",".","Returns",":","A","list","of","return","values","of","all","calls","to","`","func","(","actor",")","`","that","are","ready",".","The","values","may","be","actual","data","returned","or","exceptions","raised","during","the","remote","call","in","the","format","of","RemoteCallResults","."]
code: def fetch_ready_async_reqs(
      self,
      *,
      tags: Union[str, List[str]] = (),
      timeout_seconds: Union[None, int] = 0,
      return_obj_refs: bool = False,
      mark_healthy: bool = False,
  ) -> RemoteCallResults:
      """Get results from outstanding async requests that are ready.
  
      Automatically mark actors unhealthy if they fail to respond.
  
      Note: If tags is an empty tuple then results from all ready async requests are
      returned.
  
      Args:
          timeout_seconds: Ray.get() timeout. Default is 0 (only those that are
              already ready).
          tags: A tag or a list of tags to identify the results from this async call.
          return_obj_refs: Whether to return ObjectRef instead of actual results.
          mark_healthy: whether to mark certain actors healthy based on the results
              of these remote calls. Useful, for example, to make sure actors
              do not come back without proper state restoration.
  
      Returns:
          A list of return values of all calls to `func(actor)` that are ready.
          The values may be actual data returned or exceptions raised during the
          remote call in the format of RemoteCallResults.
      """
      # Construct the list of in-flight requests filtered by tag.
      (
          remote_calls,
          remote_actor_ids,
          valid_tags,
      ) = self.__filter_calls_by_tag(tags)
      ready, remote_results = self.__fetch_result(
          remote_actor_ids=remote_actor_ids,
          remote_calls=remote_calls,
          tags=valid_tags,
          timeout_seconds=timeout_seconds,
          return_obj_refs=return_obj_refs,
          mark_healthy=mark_healthy,
      )
  
      for obj_ref, result in zip(ready, remote_results):
          # Decrease outstanding request on this actor by 1.
          self.__remote_actor_states[
              result.actor_id
          ].num_in_flight_async_requests -= 1
          # Also, remove this call here from the in-flight list,
          # obj_refs may have already been removed when we disable an actor.
          if obj_ref in self.__in_flight_req_to_actor_id:
              del self.__in_flight_req_to_actor_id[obj_ref]
  
      return remote_results
  
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 20 for calling the FaultTolerantActorManager obj's underlying member method code with example usage: obj.fetch_ready_async_reqs() and returns: remote_results