[Core] Fix race condition in multi-threaded actor creation

python/ray/actor.py

@@ -1030,7 +1030,7 @@ def _remote(self, args=None, kwargs=None, **actor_options):
             # If this actor class was not exported in this session and job,
             # we need to export this function again, because current GCS
             # doesn't have it.
-            meta.last_export_session_and_job = worker.current_session_and_job
+
             # After serialize / deserialize modified class, the __module__
             # of modified class will be ray.cloudpickle.cloudpickle.
             # So, here pass actor_creation_function_descriptor to make
@@ -1040,6 +1040,7 @@ def _remote(self, args=None, kwargs=None, **actor_options):
                 meta.actor_creation_function_descriptor,
                 meta.method_meta.methods.keys(),
             )
+            meta.last_export_session_and_job = worker.current_session_and_job
 
         resources = ray._private.utils.resources_from_ray_options(actor_options)
         # Set the actor's default resources if not already set. First three

python/ray/tests/test_actor_advanced.py

@@ -1,6 +1,9 @@
 import os
+import random
 import sys
+import threading
 import time
+import traceback
 
 import numpy as np
 import pytest
@@ -1170,6 +1173,62 @@ def do_run(name, concurrency=4):
         assert ["ok"] * CONCURRENCY == results
 
 
+def test_create_actor_race_condition(shutdown_only):
+    """Make sure we can create actors in multiple threads without
+    race conditions.
+
+    Check https://github.com/ray-project/ray/issues/41324
+    """
+
+    @ray.remote
+    class Actor:
+        pass
+
+    def create(name, namespace, results, i):
+        time.sleep(random.random())
+        try:
+            Actor.options(
+                name=name,
+                namespace=namespace,
+                get_if_exists=True,
+                lifetime="detached",
+            ).remote()
+            results[i] = "ok"
+        except Exception:
+            e = traceback.format_exc()
+            results[i] = e
+
+    CONCURRENCY = 1000
+    ACTOR_NAME = "TestActor"
+    ACTOR_NAMESPACE = "TestNamespace"
+
+    def run_and_check():
+        results = [None] * CONCURRENCY
+        threads = [None] * CONCURRENCY
+        for i in range(CONCURRENCY):
+            threads[i] = threading.Thread(
+                target=create, args=(ACTOR_NAME, ACTOR_NAMESPACE, results, i)
+            )
+
+        for thread in threads:
+            thread.start()
+
+        for thread in threads:
+            thread.join()
+
+        for result in results:
+            assert result == "ok"
+
+        actor = ray.get_actor(
+            ACTOR_NAME, namespace=ACTOR_NAMESPACE
+        )  # Creation and get should be successful
+        ray.kill(actor)  # Cleanup
+
+    ray.init()
+    for _ in range(50):
+        run_and_check()
+
+
 def test_get_actor_in_remote_workers(ray_start_cluster):
     """Make sure we can get and create actors without
     race condition in a remote worker.