Address PR comments to change dependencies to ref count interface when applicable

This commit addresses the PR comments to change dependencies from reference_counter
to reference_counter_interface when possible to speed up build times. As a result
Some tests are modified to reflect the new change in dependecies as they depend on
the actual ReferenceCounter class.

Signed-off-by: davik <davik@anyscale.com>

Author: davik

src/ray/core_worker/BUILD.bazel

@@ -181,7 +181,7 @@ ray_cc_library(
         ":actor_handle",
         ":common",
         ":core_worker_context",
-        ":reference_counter",
+        ":reference_counter_interface",
         "//src/ray/common:id",
         "//src/ray/common:protobuf_utils",
         "//src/ray/common:task_common",
@@ -273,7 +273,7 @@ ray_cc_library(
     hdrs = ["store_provider/memory_store/memory_store.h"],
     deps = [
         ":core_worker_context",
-        ":reference_counter",
+        ":reference_counter_interface",
         "//src/ray/common:asio",
         "//src/ray/common:id",
         "//src/ray/common:ray_config",
@@ -382,7 +382,7 @@ ray_cc_library(
     hdrs = ["object_recovery_manager.h"],
     deps = [
         ":memory_store",
-        ":reference_counter",
+        ":reference_counter_interface",
         ":task_manager",
         "//src/ray/common:id",
         "//src/ray/raylet_rpc_client:raylet_client_pool",
@@ -409,7 +409,7 @@ ray_cc_library(
     deps = [
         ":common",
         ":core_worker_context",
-        ":reference_counter",
+        ":reference_counter_interface",
         "//src/ray/common:buffer",
         "//src/ray/common:id",
         "//src/ray/common:ray_config",

src/ray/core_worker/actor_manager.h

@@ -24,7 +24,7 @@
 #include "absl/container/flat_hash_map.h"
 #include "ray/core_worker/actor_creator.h"
 #include "ray/core_worker/actor_handle.h"
-#include "ray/core_worker/reference_counter.h"
+#include "ray/core_worker/reference_counter_interface.h"
 #include "ray/core_worker/task_submission/actor_task_submitter.h"
 #include "ray/gcs_rpc_client/gcs_client.h"
 namespace ray {

src/ray/core_worker/object_recovery_manager.h

@@ -22,7 +22,7 @@
 #include "absl/base/thread_annotations.h"
 #include "absl/synchronization/mutex.h"
 #include "ray/common/id.h"
-#include "ray/core_worker/reference_counter.h"
+#include "ray/core_worker/reference_counter_interface.h"
 #include "ray/core_worker/store_provider/memory_store/memory_store.h"
 #include "ray/core_worker/task_manager.h"
 #include "ray/raylet_rpc_client/raylet_client_interface.h"

src/ray/core_worker/reference_counter_interface.h

@@ -34,6 +34,7 @@ class ReferenceCounterInterface {
  public:
   using ReferenceTableProto =
       ::google::protobuf::RepeatedPtrField<rpc::ObjectReferenceCount>;
+  // Returns the amount of lineage in bytes released.
   using LineageReleasedCallback =
       std::function<int64_t(const ObjectID &, std::vector<ObjectID> *)>;
 

src/ray/core_worker/store_provider/memory_store/memory_store.h

@@ -26,7 +26,7 @@
 #include "ray/common/id.h"
 #include "ray/common/status.h"
 #include "ray/core_worker/context.h"
-#include "ray/core_worker/reference_counter.h"
+#include "ray/core_worker/reference_counter_interface.h"
 #include "ray/raylet_ipc_client/raylet_ipc_client_interface.h"
 #include "ray/rpc/utils.h"
 

src/ray/core_worker/store_provider/plasma_store_provider.h

@@ -26,7 +26,7 @@
 #include "ray/common/status.h"
 #include "ray/common/status_or.h"
 #include "ray/core_worker/context.h"
-#include "ray/core_worker/reference_counter.h"
+#include "ray/core_worker/reference_counter_interface.h"
 #include "ray/object_manager/plasma/client.h"
 #include "ray/raylet_ipc_client/raylet_ipc_client_interface.h"
 #include "src/ray/protobuf/common.pb.h"

src/ray/core_worker/tests/BUILD.bazel

@@ -51,6 +51,7 @@ ray_cc_test(
         "//src/ray/common:asio",
         "//src/ray/common:ray_object",
         "//src/ray/core_worker:memory_store",
+        "//src/ray/core_worker:reference_counter",
         "//src/ray/core_worker_rpc_client:fake_core_worker_client",
         "//src/ray/pubsub:fake_subscriber",
         "//src/ray/pubsub:publisher",
@@ -73,6 +74,7 @@ ray_cc_test(
         "//src/ray/common:test_utils",
         "//src/ray/core_worker:memory_store",
         "//src/ray/core_worker:object_recovery_manager",
+        "//src/ray/core_worker:reference_counter",
         "//src/ray/object_manager:object_manager_common",
         "//src/ray/pubsub:fake_subscriber",
         "//src/ray/raylet_rpc_client:fake_raylet_client",

src/ray/core_worker/tests/core_worker_test.cc

@@ -286,7 +286,7 @@ class CoreWorkerTest : public ::testing::Test {
   boost::thread io_thread_;
 
   rpc::Address rpc_address_;
-  std::shared_ptr<ReferenceCounter> reference_counter_;
+  std::shared_ptr<ReferenceCounterInterface> reference_counter_;
   std::shared_ptr<CoreWorkerMemoryStore> memory_store_;
   ActorTaskSubmitter *actor_task_submitter_;
   pubsub::Publisher *object_info_publisher_;
@@ -537,9 +537,10 @@ TEST_F(CoreWorkerTest, HandleGetObjectStatusObjectOutOfScope) {
 
 namespace {
 
-ObjectID CreateInlineObjectInMemoryStoreAndRefCounter(CoreWorkerMemoryStore &memory_store,
-                                                      ReferenceCounter &reference_counter,
-                                                      rpc::Address &rpc_address) {
+ObjectID CreateInlineObjectInMemoryStoreAndRefCounter(
+    CoreWorkerMemoryStore &memory_store,
+    ReferenceCounterInterface &reference_counter,
+    rpc::Address &rpc_address) {
   auto inlined_dependency_id = ObjectID::FromRandom();
   std::string data = "hello";
   auto data_ptr = const_cast<uint8_t *>(reinterpret_cast<const uint8_t *>(data.data()));

src/ray/core_worker/tests/object_recovery_manager_test.cc

@@ -25,6 +25,7 @@
 #include "mock/ray/core_worker/task_manager_interface.h"
 #include "mock/ray/pubsub/publisher.h"
 #include "ray/common/test_utils.h"
+#include "ray/core_worker/reference_counter.h"
 #include "ray/core_worker/store_provider/memory_store/memory_store.h"
 #include "ray/pubsub/fake_subscriber.h"
 #include "ray/raylet_rpc_client/fake_raylet_client.h"
@@ -181,7 +182,7 @@ class ObjectRecoveryManagerTestBase : public ::testing::Test {
   std::shared_ptr<rpc::RayletClientPool> raylet_client_pool_;
   std::shared_ptr<MockRayletClient> raylet_client_;
   std::shared_ptr<MockTaskManager> task_manager_;
-  std::shared_ptr<ReferenceCounter> ref_counter_;
+  std::shared_ptr<ReferenceCounterInterface> ref_counter_;
   ObjectRecoveryManager manager_;
 };
 

src/ray/core_worker/tests/reference_counter_test.cc

@@ -41,7 +41,7 @@ static const ReferenceCounterInterface::ReferenceTableProto empty_refs;
 
 class ReferenceCountTest : public ::testing::Test {
  protected:
-  std::unique_ptr<ReferenceCounter> rc;
+  std::unique_ptr<ReferenceCounterInterface> rc;
   virtual void SetUp() {
     rpc::Address addr;
     publisher_ = std::make_shared<pubsub::MockPublisher>();
@@ -67,7 +67,7 @@ class ReferenceCountTest : public ::testing::Test {
 
 class ReferenceCountLineageEnabledTest : public ::testing::Test {
  protected:
-  std::unique_ptr<ReferenceCounter> rc;
+  std::unique_ptr<ReferenceCounterInterface> rc;
   virtual void SetUp() {
     rpc::Address addr;
     publisher_ = std::make_shared<pubsub::MockPublisher>();