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: Try fetching results from remote actor calls.
  
  Mark whether an actor is healthy or not accordingly.
  
  Args:
      remote_actor_ids: IDs of the actors these remote
          calls were fired against.
      remote_calls: list of remote calls to fetch.
      tags: list of tags used for identifying the remote calls.
      timeout_seconds: timeout for the ray.wait() call. Default is None.
      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 ready ObjectRefs mapping to the results of those calls.
docstr_tok: ["Try","fetching","results","from","remote","actor","calls",".","Mark","whether","an","actor","is","healthy","or","not","accordingly",".","Args",":","remote_actor_ids",":","IDs","of","the","actors","these","remote","calls","were","fired","against",".","remote_calls",":","list","of","remote","calls","to","fetch",".","tags",":","list","of","tags","used","for","identifying","the","remote","calls",".","timeout_seconds",":","timeout","for","the","ray.wait","(",")","call",".","Default","is","None",".","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","ready","ObjectRefs","mapping","to","the","results","of","those","calls","."]
code: def __fetch_result(
      self,
      *,
      remote_actor_ids: List[int],
      remote_calls: List[ray.ObjectRef],
      tags: List[str],
      timeout_seconds: int = None,
      return_obj_refs: bool = False,
      mark_healthy: bool = False,
  ) -> Tuple[List[ray.ObjectRef], RemoteCallResults]:
      """Try fetching results from remote actor calls.
  
      Mark whether an actor is healthy or not accordingly.
  
      Args:
          remote_actor_ids: IDs of the actors these remote
              calls were fired against.
          remote_calls: list of remote calls to fetch.
          tags: list of tags used for identifying the remote calls.
          timeout_seconds: timeout for the ray.wait() call. Default is None.
          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 ready ObjectRefs mapping to the results of those calls.
      """
      # Notice that we do not return the refs to any unfinished calls to the
      # user, since it is not safe to handle such remote actor calls outside the
      # context of this actor manager. These requests are simply dropped.
      timeout = (
          float(timeout_seconds)
          if timeout_seconds is not None
          else None
      )
  
      # This avoids calling ray.init() in the case of 0 remote calls.
      # This is useful if the number of remote workers is 0.
      if not remote_calls:
          return [], RemoteCallResults()
  
      ready, _ = ray.wait(
          remote_calls,
          num_returns=len(remote_calls),
          timeout=timeout,
          # Make sure remote results are fetched locally in parallel.
          fetch_local=not return_obj_refs,
      )
  
      # Remote data should already be fetched to local object store at this point.
      remote_results = RemoteCallResults()
      for r in ready:
          # Find the corresponding actor ID for this remote call.
          actor_id = remote_actor_ids[remote_calls.index(r)]
          tag = tags[remote_calls.index(r)]
  
          # If caller wants ObjectRefs, return directly without resolve them.
          if return_obj_refs:
              remote_results.add_result(
                  actor_id, ResultOrError(result=r), tag
              )
              continue
  
          try:
              result = ray.get(r)
              remote_results.add_result(
                  actor_id, ResultOrError(result=result), tag
              )
  
              # Actor came back from an unhealthy state. Mark this actor as healthy
              # and add it to our restored set.
              if mark_healthy and not self.is_actor_healthy(actor_id):
                  logger.info(
                      f"brining actor {actor_id} back into service."
                  )
                  self.set_actor_state(actor_id, healthy=True)
                  self._num_actor_restarts += 1
          except Exception as e:
              # Return error to the user.
              remote_results.add_result(
                  actor_id, ResultOrError(error=e), tag
              )
  
              # Mark the actor as unhealthy.
              # TODO(jungong): Using RayError here to preserve historical behavior.
              #  It may very likely be better to use RayActorError here.
              if isinstance(e, RayError):
                  # Take this actor out of service and wait for Ray Core to
                  # restore it.
                  if self.is_actor_healthy(actor_id):
                      logger.error(
                          f"Ray error, taking actor {actor_id} out of service. "
                          f"{str(e)}"
                      )
                  self.set_actor_state(actor_id, healthy=False)
              else:
                  # ActorManager should not handle application level errors.
                  pass
  
      return ready, 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 15 for calling the FaultTolerantActorManager obj's underlying member method code with example usage: obj.__fetch_result() and returns: ready,  remote_results, list,  RemoteCallResults