[Core] Remove reference counter mock for real reference counter in testing

src/ray/core_worker/actor_manager.cc

@@ -80,10 +80,10 @@ std::pair<std::shared_ptr<const ActorHandle>, Status> ActorManager::GetNamedActo
   if (status.ok()) {
     auto actor_handle = std::make_unique<ActorHandle>(actor_table_data, task_spec);
     actor_id = actor_handle->GetActorID();
-    AddNewActorHandle(std::move(actor_handle),
-                      call_site,
-                      caller_address,
-                      /*owned*/ false);
+    EmplaceNewActorHandle(std::move(actor_handle),
+                          call_site,
+                          caller_address,
+                          /*owned*/ false);
   } else {
     // Use a NIL actor ID to signal that the actor wasn't found.
     RAY_LOG(DEBUG) << "Failed to look up actor with name: " << name;
@@ -129,16 +129,24 @@ std::shared_ptr<ActorHandle> ActorManager::GetActorHandleIfExists(
   return nullptr;
 }
 
-bool ActorManager::AddNewActorHandle(std::unique_ptr<ActorHandle> actor_handle,
-                                     const std::string &call_site,
-                                     const rpc::Address &caller_address,
-                                     bool owned) {
+bool ActorManager::EmplaceNewActorHandle(std::unique_ptr<ActorHandle> actor_handle,
+                                         const std::string &call_site,
+                                         const rpc::Address &caller_address,
+                                         bool owned) {
   const auto &actor_id = actor_handle->GetActorID();
   const auto actor_creation_return_id = ObjectID::ForActorHandle(actor_id);
+
+  {
+    absl::MutexLock lock(&mutex_);
+    if (actor_handles_.contains(actor_id)) {
+      return false;
+    }
+  }
+
   // Detached actor doesn't need ref counting.
   if (owned) {
     reference_counter_.AddOwnedObject(actor_creation_return_id,
-                                      /*inner_ids=*/{},
+                                      /*contained_ids=*/{},
                                       caller_address,
                                       call_site,
                                       /*object_size*/ -1,

src/ray/core_worker/actor_manager.h

@@ -106,6 +106,7 @@ class ActorManager {
   ///
   /// NOTE: Getting an actor handle from GCS (named actor) is considered as adding a new
   /// actor handle.
+  /// NOTE: Attempting to add the same actor in parallel can cause RAY CHECK failure.
   ///
   /// \param actor_handle The handle to the actor.
   /// \param[in] call_site The caller's site.
@@ -114,10 +115,10 @@ class ActorManager {
   /// task.
   /// \return True if the handle was added and False if we already had a handle to
   /// the same actor.
-  bool AddNewActorHandle(std::unique_ptr<ActorHandle> actor_handle,
-                         const std::string &call_site,
-                         const rpc::Address &caller_address,
-                         bool owned);
+  bool EmplaceNewActorHandle(std::unique_ptr<ActorHandle> actor_handle,
+                             const std::string &call_site,
+                             const rpc::Address &caller_address,
+                             bool owned);
 
   /// Wait for actor reference deletion.
   ///

src/ray/core_worker/core_worker.cc

@@ -2164,9 +2164,11 @@ Status CoreWorker::CreateActor(const RayFunction &function,
   // Add the actor handle before we submit the actor creation task, since the
   // actor handle must be in scope by the time the GCS sends the
   // WaitForActorRefDeletedRequest.
-  RAY_CHECK(actor_manager_->AddNewActorHandle(
+  RAY_CHECK(actor_manager_->EmplaceNewActorHandle(
       std::move(actor_handle), CurrentCallSite(), rpc_address_, /*owned=*/!is_detached))
-      << "Actor " << actor_id << " already exists";
+      << "Attempt to emplace new actor handle for the actor being created with actor id: "
+      << actor_id
+      << " failed because an actor handle with the same actor id has already been added";
   *return_actor_id = actor_id;
   TaskSpecification task_spec = std::move(builder).ConsumeAndBuild();
   RAY_LOG(DEBUG) << "Submitting actor creation task " << task_spec.DebugString();

src/ray/core_worker/core_worker.h

@@ -865,6 +865,9 @@ class CoreWorker {
 
   /// Create an actor.
   ///
+  /// NOTE: RAY CHECK fails if an actor handle with the same actor id has already been
+  /// added, or if the scheduling strategy for actor creation is not set.
+  ///
   /// \param[in] caller_id ID of the task submitter.
   /// \param[in] function The remote function that generates the actor object.
   /// \param[in] args Arguments of this task.

src/ray/core_worker/reference_counter_interface.h

@@ -125,6 +125,8 @@ class ReferenceCounterInterface {
   /// owner ID will change for workers executing normal tasks and it is
   /// possible to have leftover references after a task has finished.
   ///
+  /// NOTE: RAY CHECK fails if the object was already added.
+  ///
   /// \param[in] object_id The ID of the object that we own.
   /// \param[in] contained_ids ObjectIDs that are contained in the object's value.
   /// As long as the object_id is in scope, the inner objects should not be GC'ed.

src/ray/core_worker/task_submission/tests/BUILD.bazel

@@ -34,7 +34,10 @@ ray_cc_test(
     tags = ["team:core"],
     deps = [
         "//:ray_mock",
+        "//src/ray/core_worker:reference_counter",
         "//src/ray/core_worker/task_submission:actor_task_submitter",
+        "//src/ray/pubsub:fake_publisher",
+        "//src/ray/pubsub:fake_subscriber",
         "@com_google_googletest//:gtest",
         "@com_google_googletest//:gtest_main",
     ],
@@ -54,6 +57,8 @@ ray_cc_test(
         "//src/ray/core_worker:reference_counter",
         "//src/ray/core_worker:task_manager",
         "//src/ray/core_worker_rpc_client:fake_core_worker_client",
+        "//src/ray/pubsub:fake_publisher",
+        "//src/ray/pubsub:fake_subscriber",
         "@com_google_googletest//:gtest",
         "@com_google_googletest//:gtest_main",
     ],