docstr: Calls the given function with this Actor instance.
  
  A generic interface for applying arbitray member functions on a
  remote actor.
  
  Args:
      func: The function to call, with this RolloutWorker as first
          argument, followed by args, and kwargs.
      args: Optional additional args to pass to the function call.
      kwargs: Optional additional kwargs to pass to the function call.
  
  Returns:
      The return value of the function call.
docstr_tok: ["Calls","the","given","function","with","this","Actor","instance",".","A","generic","interface","for","applying","arbitray","member","functions","on","a","remote","actor",".","Args",":","func",":","The","function","to","call",",","with","this","RolloutWorker","as","first","argument",",","followed","by","args",",","and","kwargs",".","args",":","Optional","additional","args","to","pass","to","the","function","call",".","kwargs",":","Optional","additional","kwargs","to","pass","to","the","function","call",".","Returns",":","The","return","value","of","the","function","call","."]
code: def apply(
      self,
      func: Callable[[Any, Optional[Any], Optional[Any]], T],
      *args,
      **kwargs,
  ) -> T:
      """Calls the given function with this Actor instance.
  
      A generic interface for applying arbitray member functions on a
      remote actor.
  
      Args:
          func: The function to call, with this RolloutWorker as first
              argument, followed by args, and kwargs.
          args: Optional additional args to pass to the function call.
          kwargs: Optional additional kwargs to pass to the function call.
  
      Returns:
          The return value of the function call.
      """
      try:
          return func(self, *args, **kwargs)
      except Exception as e:
          # Actor should be recreated by Ray.
          if self.config.recreate_failed_workers:
              logger.exception(
                  f"Worker exception caught during `apply()`: {e}"
              )
              # Small delay to allow logs messages to propagate.
              time.sleep(self.config.delay_between_worker_restarts_s)
              # Kill this worker so Ray Core can restart it.
              sys.exit(1)
          # Actor should be left dead.
          else:
              raise e
  
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 2 for calling the FaultAwareApply obj's underlying member method code with example usage: obj.apply(func) and returns: func