Register function race (#9346)

Author: wuisawesome

python/ray/import_thread.py

@@ -136,9 +136,13 @@ def _process_key(self, key):
                         ray_constants.DUPLICATE_REMOTE_FUNCTION_THRESHOLD)
 
         if key.startswith(b"RemoteFunction"):
-            with profiling.profile("register_remote_function"):
-                (self.worker.function_actor_manager.
-                 fetch_and_register_remote_function(key))
+            # TODO (Alex): There's a race condition here if the worker is
+            # shutdown before the function finished registering (because core
+            # worker's global worker is unset before shutdown and is needed
+            # for profiling).
+            # with profiling.profile("register_remote_function"):
+            (self.worker.function_actor_manager.
+             fetch_and_register_remote_function(key))
         elif key.startswith(b"FunctionsToRun"):
             with profiling.profile("fetch_and_run_function"):
                 self.fetch_and_execute_function_to_run(key)

python/ray/remote_function.py

@@ -41,7 +41,7 @@ class RemoteFunction:
         _num_return_vals: The default number of return values for invocations
             of this remote function.
         _max_calls: The number of times a worker can execute this function
-            before executing.
+            before exiting.
         _decorator: An optional decorator that should be applied to the remote
             function invocation (as opposed to the function execution) before
             invoking the function. The decorator must return a function that

python/ray/tests/test_advanced.py

@@ -202,7 +202,8 @@ def f():
             "ray.wait",
             "submit_task",
             "fetch_and_run_function",
-            "register_remote_function",
+            # TODO (Alex) :https://github.com/ray-project/ray/pull/9346
+            # "register_remote_function",
             "custom_event",  # This is the custom one from ray.profile.
         ]
 

src/ray/core_worker/core_worker.cc

@@ -149,10 +149,6 @@ void CoreWorkerProcess::EnsureInitialized() {
 CoreWorker &CoreWorkerProcess::GetCoreWorker() {
   EnsureInitialized();
   if (instance_->options_.num_workers == 1) {
-    // TODO(mehrdadn): Remove this when the bug is resolved.
-    // Somewhat consistently reproducible via
-    // python/ray/tests/test_basic.py::test_background_tasks_with_max_calls
-    // with -c opt on Windows.
     RAY_CHECK(instance_->global_worker_) << "global_worker_ must not be NULL";
     return *instance_->global_worker_;
   }