[core] remove import thread (ray-project#36293)

Why are these changes needed?
import thread is no longer needed, remove it.

python/ray/_private/function_manager.py

@@ -154,9 +154,6 @@ def export_key(self, key):
  # One optimization is that we can use importer counter since
  # it's sure keys before this counter has been allocated.
  with self._export_lock:
- self._num_exported = max(
- self._num_exported, self._worker.import_thread.num_imported
- )
  while True:
  self._num_exported += 1
  holder = make_export_key(

python/ray/_private/worker.py

@@ -43,7 +43,6 @@
  from typing_extensions import Literal, Protocol
 
 import ray
-import ray._private.import_thread as import_thread
 import ray._private.node
 import ray._private.parameter
 import ray._private.profiling as profiling
@@ -2248,21 +2247,6 @@ def connect(
  " and will be removed in the future."
  )
 
- # Setup import thread and start the import thread
- # if the worker has job_id initialized.
- # Otherwise, defer the start up of
- # import thread until job_id is initialized.
- # (python/ray/_raylet.pyx maybe_initialize_job_config)
- if mode not in (RESTORE_WORKER_MODE, SPILL_WORKER_MODE):
- worker.import_thread = import_thread.ImportThread(
- worker, mode, worker.threads_stopped
- )
- if (
- worker.current_job_id != JobID.nil()
- and ray._raylet.Config.start_python_importer_thread()
- ):
- worker.import_thread.start()
-
  # If this is a driver running in SCRIPT_MODE, start a thread to print error
  # messages asynchronously in the background. Ideally the scheduler would
  # push messages to the driver's worker service, but we ran into bugs when
@@ -2317,8 +2301,6 @@ def disconnect(exiting_interpreter=False):
  worker.gcs_error_subscriber.close()
  if hasattr(worker, "gcs_log_subscriber"):
  worker.gcs_log_subscriber.close()
- if hasattr(worker, "import_thread"):
- worker.import_thread.join_import_thread()
  if hasattr(worker, "listener_thread"):
  worker.listener_thread.join()
  if hasattr(worker, "logger_thread"):

python/ray/_raylet.pyx

@@ -2102,9 +2102,6 @@ def maybe_initialize_job_config():
  print(job_id_magic_token, end="")
  print(job_id_magic_token, file=sys.stderr, end="")
 
- # Only start import thread after job_config is initialized
- ray._private.worker.start_import_thread()
-
  job_config_initialized = True
 
 

python/ray/tests/test_basic_5.py

@@ -208,10 +208,9 @@ def get_kv_metrics():
  b'cluster' b'CLUSTER_METADATA'
  b'tracing' b'tracing_startup_hook'
  b'fun' b'IsolatedExports:01000000:\x00\x00\x00\x00\x00\x00\x00\x01'
- b'fun' b'RemoteFunction:01000000:'
  """
  # !!!If you want to increase this number, please let ray-core knows this!!!
- assert freqs["internal_kv_get"] == 4
+ assert freqs["internal_kv_get"] == 3
 
 
 @pytest.mark.skipif(sys.platform == "win32", reason="Fails on Windows.")