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: Filter out func and remote worker ids by actor state.
  
  Args:
      func: A single, or a list of Callables.
      remote_actor_ids: IDs of potential remote workers to apply func on.
  
  Returns:
      A tuple of (filtered func, filtered remote worker ids).
docstr_tok: ["Filter","out","func","and","remote","worker","ids","by","actor","state",".","Args",":","func",":","A","single",",","or","a","list","of","Callables",".","remote_actor_ids",":","IDs","of","potential","remote","workers","to","apply","func","on",".","Returns",":","A","tuple","of","(","filtered","func",",","filtered","remote","worker","ids",")","."]
code: def _filter_func_and_remote_actor_id_by_state(
      self,
      func: Union[Callable[[Any], Any], List[Callable[[Any], Any]]],
      remote_actor_ids: List[int],
  ):
      """Filter out func and remote worker ids by actor state.
  
      Args:
          func: A single, or a list of Callables.
          remote_actor_ids: IDs of potential remote workers to apply func on.
  
      Returns:
          A tuple of (filtered func, filtered remote worker ids).
      """
      if isinstance(func, list):
          assert len(remote_actor_ids) == len(
              func
          ), "Func must have the same number of callables as remote actor ids."
          # We are given a list of functions to apply.
          # Need to filter the functions together with worker IDs.
          temp_func = []
          temp_remote_actor_ids = []
          for f, i in zip(func, remote_actor_ids):
              if self.is_actor_healthy(i):
                  temp_func.append(f)
                  temp_remote_actor_ids.append(i)
          func = temp_func
          remote_actor_ids = temp_remote_actor_ids
      else:
          # Simply filter the worker IDs.
          remote_actor_ids = [
              i for i in remote_actor_ids if self.is_actor_healthy(i)
          ]
  
      return func, remote_actor_ids
  
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 16 for calling the FaultTolerantActorManager obj's underlying member method code with example usage: obj._filter_func_and_remote_actor_id_by_state(func, remote_actor_ids) and returns: func,  remote_actor_ids