[Core] Remove reference counter mock for real reference counter in testing (#57178)

This commit takes a step towards utilizing real classes during unit
testing instead of mocks for components that are simple and local (e.g.
reference counter). We remove the reference counter mock as promised in
#57177 (review).

By directly testing on the real reference_counter, we are able to catch
a behavior mismatch between AddNewActorHandle and its doc described
behavior. Specifically, the EmplaceNewActorHandle doc states that the
function should return false instead of failing due to ray check and the
test tests exactly that. However, due to calling AddOwnedObject,
EmplaceNewActorHandle actually ray check fails. The commit addresses
this issue by making AddNewActorHandle idempotent.

---------

Signed-off-by: davik <davik@anyscale.com>
Signed-off-by: Kunchen (David) Dai <54918178+Kunchd@users.noreply.github.com>
Co-authored-by: davik <davik@anyscale.com>
Co-authored-by: Ibrahim Rabbani <irabbani@anyscale.com>
Co-authored-by: Ibrahim Rabbani <israbbani@gmail.com>

Author: 3 people

src/mock/ray/core_worker/reference_counter.h

@@ -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.cc

@@ -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/actor_manager.h

@@ -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.cc

@@ -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/core_worker.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/reference_counter_interface.h

@@ -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",
     ],