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: Wait until the function to be executed is present on this worker.
  This method will simply loop until the import thread has imported the
  relevant function. If we spend too long in this loop, that may indicate
  a problem somewhere and we will push an error message to the user.
  If this worker is an actor, then this will wait until the actor has
  been defined.
  Args:
      function_descriptor : The FunctionDescriptor of the function that
          we want to execute.
      job_id: The ID of the job to push the error message to
          if this times out.
code: def _wait_for_function(
      self, function_descriptor, job_id: str, timeout=10
  ):
      """Wait until the function to be executed is present on this worker.
      This method will simply loop until the import thread has imported the
      relevant function. If we spend too long in this loop, that may indicate
      a problem somewhere and we will push an error message to the user.
      If this worker is an actor, then this will wait until the actor has
      been defined.
      Args:
          function_descriptor : The FunctionDescriptor of the function that
              we want to execute.
          job_id: The ID of the job to push the error message to
              if this times out.
      """
      start_time = time.time()
      # Only send the warning once.
      warning_sent = False
      while True:
          with self.lock:
              if self._worker.actor_id.is_nil():
                  if (
                      function_descriptor.function_id
                      in self._function_execution_info
                  ):
                      break
                  else:
                      key = make_function_table_key(
                          b"RemoteFunction",
                          job_id,
                          function_descriptor.function_id.binary(),
                      )
                      if (
                          self.fetch_and_register_remote_function(key)
                          is True
                      ):
                          break
              else:
                  assert not self._worker.actor_id.is_nil()
                  # Actor loading will happen when execute_task is called.
                  assert self._worker.actor_id in self._worker.actors
                  break
  
          if time.time() - start_time > timeout:
              warning_message = (
                  "This worker was asked to execute a function "
                  f"that has not been registered ({function_descriptor}, "
                  f"node={self._worker.node_ip_address}, "
                  f"worker_id={self._worker.worker_id.hex()}, "
                  f"pid={os.getpid()}). You may have to restart Ray."
              )
              if not warning_sent:
                  logger.error(warning_message)
                  ray._private.utils.push_error_to_driver(
                      self._worker,
                      ray_constants.WAIT_FOR_FUNCTION_PUSH_ERROR,
                      warning_message,
                      job_id=job_id,
                  )
              warning_sent = True
          # Try importing in case the worker did not get notified, or the
          # importer thread did not run.
          self._worker.import_thread._do_importing()
          time.sleep(0.001)
  
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 13 for calling the FunctionActorManager obj's underlying member method code with example usage: obj._wait_for_function(function_descriptor, job_id, timeout) without return types