[core]Remove the legacy code related to import thread.

python/ray/_private/function_manager.py

@@ -475,9 +475,6 @@ def _wait_for_function(self, function_descriptor, job_id: str, timeout=10):
                         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)
 
     def export_actor_class(

python/ray/_private/worker.py

@@ -2507,19 +2507,6 @@ def disconnect(exiting_interpreter=False):
         ray_actor._ActorClassMethodMetadata.reset_cache()
 
 
-def start_import_thread():
-    """Start the import thread if the worker is connected."""
-    worker = global_worker
-    worker.check_connected()
-
-    assert _mode() not in (
-        RESTORE_WORKER_MODE,
-        SPILL_WORKER_MODE,
-    ), "import thread can not be used in IO workers."
-    if worker.import_thread and ray._raylet.Config.start_python_importer_thread():
-        worker.import_thread.start()
-
-
 @contextmanager
 def _changeproctitle(title, next_title):
     if _mode() is not LOCAL_MODE:

python/ray/includes/ray_config.pxd

@@ -61,8 +61,6 @@ cdef extern from "ray/common/ray_config.h" nogil:
 
         c_bool gcs_grpc_based_pubsub() const
 
-        c_bool start_python_importer_thread() const
-
         c_string REDIS_CA_CERT() const
 
         c_string REDIS_CA_PATH() const

python/ray/includes/ray_config.pxi

@@ -97,10 +97,6 @@ cdef class Config:
     def record_ref_creation_sites():
         return RayConfig.instance().record_ref_creation_sites()
 
-    @staticmethod
-    def start_python_importer_thread():
-        return RayConfig.instance().start_python_importer_thread()
-
     @staticmethod
     def REDIS_CA_CERT():
         return RayConfig.instance().REDIS_CA_CERT()

src/ray/common/ray_config_def.h

@@ -427,11 +427,6 @@ RAY_CONFIG(uint32_t, task_oom_retry_delay_base_ms, 1000)
 /// Duration to wait between retrying to kill a task.
 RAY_CONFIG(uint32_t, cancellation_retry_ms, 2000)
 
-/// Whether to start a background thread to import Python dependencies eagerly.
-/// When set to false, Python dependencies will still be imported, only when
-/// they are needed.
-RAY_CONFIG(bool, start_python_importer_thread, true)
-
 /// Determines if forking in Ray actors / tasks are supported.
 /// Note that this only enables forking in workers, but not drivers.
 RAY_CONFIG(bool, support_fork, false)

src/ray/raylet/worker_pool.cc

@@ -422,8 +422,6 @@ WorkerPool::BuildProcessCommandArgs(const Language &language,
     // Support forking in gRPC.
     env.insert({"GRPC_ENABLE_FORK_SUPPORT", "True"});
     env.insert({"GRPC_POLL_STRATEGY", "poll"});
-    // Make sure only the main thread is running in Python workers.
-    env.insert({"RAY_start_python_importer_thread", "0"});
   }
 
   return {std::move(worker_command_args), std::move(env)};