class_docstr: A class used to export/load remote functions and actors.
  Attributes:
      _worker: The associated worker that this manager related.
      _functions_to_export: The remote functions to export when
          the worker gets connected.
      _actors_to_export: The actors to export when the worker gets
          connected.
      _function_execution_info: The function_id
          and execution_info.
      _num_task_executions: The function
          execution times.
      imported_actor_classes: The set of actor classes keys (format:
          ActorClass:function_id) that are already in GCS.
class_docstr_tok: ["A","class","used","to","export\/load","remote","functions","and","actors",".","Attributes",":","_worker",":","The","associated","worker","that","this","manager","related",".","_functions_to_export",":","The","remote","functions","to","export","when","the","worker","gets","connected",".","_actors_to_export",":","The","actors","to","export","when","the","worker","gets","connected",".","_function_execution_info",":","The","function_id","and","execution_info",".","_num_task_executions",":","The","function","execution","times",".","imported_actor_classes",":","The","set","of","actor","classes","keys","(","format",":","ActorClass",":","function_id",")","that","are","already","in","GCS","."]
docstr: Import a remote function.
code: def fetch_and_register_remote_function(self, key):
      """Import a remote function."""
      remote_function_info = self.fetch_registered_method(key)
      if not remote_function_info:
          return False
      (
          job_id_str,
          function_id_str,
          function_name,
          serialized_function,
          module,
          max_calls,
      ) = remote_function_info
  
      function_id = ray.FunctionID(function_id_str)
      job_id = ray.JobID(job_id_str)
      max_calls = int(max_calls)
  
      # This function is called by ImportThread. This operation needs to be
      # atomic. Otherwise, there is race condition. Another thread may use
      # the temporary function above before the real function is ready.
      with self.lock:
          self._num_task_executions[function_id] = 0
  
          try:
              function = pickle.loads(serialized_function)
          except Exception:
              # If an exception was thrown when the remote function was
              # imported, we record the traceback and notify the scheduler
              # of the failure.
              traceback_str = format_error_message(
                  traceback.format_exc()
              )
  
              def f(*args, **kwargs):
                  raise RuntimeError(
                      "The remote function failed to import on the "
                      "worker. This may be because needed library "
                      "dependencies are not installed in the worker "
                      "environment or cannot be found from sys.path "
                      f"{sys.path}:\n\n{traceback_str}"
                  )
  
              # Use a placeholder method when function pickled failed
              self._function_execution_info[
                  function_id
              ] = FunctionExecutionInfo(
                  function=f,
                  function_name=function_name,
                  max_calls=max_calls,
              )
  
              # Log the error message. Log at DEBUG level to avoid overly
              # spamming the log on import failure. The user gets the error
              # via the RuntimeError message above.
              logger.debug(
                  "Failed to unpickle the remote function "
                  f"'{function_name}' with "
                  f"function ID {function_id.hex()}. "
                  f"Job ID:{job_id}."
                  f"Traceback:\n{traceback_str}. "
              )
          else:
              # The below line is necessary. Because in the driver process,
              # if the function is defined in the file where the python
              # script was started from, its module is `__main__`.
              # However in the worker process, the `__main__` module is a
              # different module, which is `default_worker.py`
              function.__module__ = module
              self._function_execution_info[
                  function_id
              ] = FunctionExecutionInfo(
                  function=function,
                  function_name=function_name,
                  max_calls=max_calls,
              )
      return True
  
imports: import dis
  import sys
  import hashlib
  import importlib
  import inspect
  import json
  import logging
  import os
  import threading
  import time
  import traceback
  from collections import defaultdict, namedtuple
  from typing import Optional, Callable
  import ray
  import ray._private.profiling
  from ray import cloudpickle
  from ray._private import ray_constants
  from ray._private.inspect_util import is_class_method, is_function_or_method, is_static_method
  from ray._private.ray_constants import KV_NAMESPACE_FUNCTION_TABLE
  from ray._private.utils import check_oversized_function, ensure_str, format_error_message
  from ray._private.serialization import pickle_dumps
  from ray._raylet import JobID, PythonFunctionDescriptor, WORKER_PROCESS_SETUP_HOOK_KEY_NAME_GCS
label_desc: Use image node_id 10 for calling the FunctionActorManager obj's underlying member method code with example usage: obj.fetch_and_register_remote_function(key) and returns: True, False