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: Ping all unhealthy actors to try bringing them back.
  
  Args:
      timeout_seconds: Timeout to avoid pinging hanging workers indefinitely.
      mark_healthy: Whether to mark actors healthy if they respond to the ping.
  
  Returns:
      A list of actor IDs that were restored by the `ping` AND those actors that
      were previously restored via other remote requests. The cached set of
      such previously restored actors will be erased in this call.
docstr_tok: ["Ping","all","unhealthy","actors","to","try","bringing","them","back",".","Args",":","timeout_seconds",":","Timeout","to","avoid","pinging","hanging","workers","indefinitely",".","mark_healthy",":","Whether","to","mark","actors","healthy","if","they","respond","to","the","ping",".","Returns",":","A","list","of","actor","IDs","that","were","restored","by","the","`","ping","`","AND","those","actors","that","were","previously","restored","via","other","remote","requests",".","The","cached","set","of","such","previously","restored","actors","will","be","erased","in","this","call","."]
code: def probe_unhealthy_actors(
      self,
      timeout_seconds: Optional[int] = None,
      mark_healthy: bool = False,
  ) -> List[int]:
      """Ping all unhealthy actors to try bringing them back.
  
      Args:
          timeout_seconds: Timeout to avoid pinging hanging workers indefinitely.
          mark_healthy: Whether to mark actors healthy if they respond to the ping.
  
      Returns:
          A list of actor IDs that were restored by the `ping` AND those actors that
          were previously restored via other remote requests. The cached set of
          such previously restored actors will be erased in this call.
      """
      # Collect recently restored actors (from `self.__fetch_result` calls other than
      # the one triggered here via the `ping`).
      restored_actors = list(self.__restored_actors)
      self.__restored_actors.clear()
  
      # Probe all unhealthy actors via a simple `ping()`.
      unhealthy_actor_ids = [
          actor_id
          for actor_id in self.actor_ids()
          if not self.is_actor_healthy(actor_id)
      ]
      # No unhealthy actors currently -> Return recently restored ones.
      if not unhealthy_actor_ids:
          return restored_actors
  
      # Some unhealthy actors -> `ping()` all of them to trigger a new fetch and
      # capture all restored ones.
      remote_results = self.foreach_actor(
          func=lambda actor: actor.ping(),
          remote_actor_ids=unhealthy_actor_ids,
          healthy_only=False,  # We specifically want to ping unhealthy actors.
          timeout_seconds=timeout_seconds,
          mark_healthy=mark_healthy,
      )
  
      # Return previously restored actors AND actors restored via the `ping()` call.
      return restored_actors + [
          result.actor_id for result in remote_results if result.ok
      ]
  
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 21 for calling the FaultTolerantActorManager obj's underlying member method code with example usage: obj.probe_unhealthy_actors(timeout_seconds, mark_healthy) and returns: unknown, restored_actors