[GCS]Limit the number of profile table (#10888)

* add part code

* add part code

* fix compile bug

* fix compile bug

* fix review comments

Co-authored-by: 灵洵 <fengbin.ffb@antfin.com>

src/ray/common/ray_config_def.h

@@ -208,6 +208,9 @@ RAY_CONFIG(uint32_t, maximum_gcs_deletion_batch_size, 1000)
 /// Maximum number of items in one batch to scan/get/delete from GCS storage.
 RAY_CONFIG(uint32_t, maximum_gcs_storage_operation_batch_size, 1000)
 
+/// Maximum number of rows in GCS profile table.
+RAY_CONFIG(int32_t, maximum_profile_table_rows_count, 10 * 1000)
+
 /// When getting objects from object store, print a warning every this number of attempts.
 RAY_CONFIG(uint32_t, object_store_get_warn_per_num_attempts, 50)
 

src/ray/gcs/gcs_client/test/global_state_accessor_test.cc

@@ -192,7 +192,7 @@ TEST_F(GlobalStateAccessorTest, TestInternalConfig) {
 }
 
 TEST_F(GlobalStateAccessorTest, TestProfileTable) {
-  int profile_count = 100;
+  int profile_count = RayConfig::instance().maximum_profile_table_rows_count() + 1;
   ASSERT_EQ(global_state_->GetAllProfileInfo().size(), 0);
   for (int index = 0; index < profile_count; ++index) {
     auto client_id = ClientID::FromRandom();
@@ -203,7 +203,8 @@ TEST_F(GlobalStateAccessorTest, TestProfileTable) {
         [&promise](Status status) { promise.set_value(status.ok()); }));
     WaitReady(promise.get_future(), timeout_ms_);
   }
-  ASSERT_EQ(global_state_->GetAllProfileInfo().size(), profile_count);
+  ASSERT_EQ(global_state_->GetAllProfileInfo().size(),
+            RayConfig::instance().maximum_profile_table_rows_count());
 }
 
 TEST_F(GlobalStateAccessorTest, TestObjectTable) {

src/ray/gcs/gcs_server/stats_handler_impl.cc

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 #include "ray/gcs/gcs_server/stats_handler_impl.h"
+#include "ray/common/ray_config.h"
 
 namespace ray {
 namespace rpc {
@@ -34,7 +35,12 @@ void DefaultStatsHandler::HandleAddProfileData(const AddProfileDataRequest &requ
     GCS_RPC_SEND_REPLY(send_reply_callback, reply, status);
   };
 
-  Status status = gcs_table_storage_->ProfileTable().Put(UniqueID::FromRandom(),
+  // To limit the number of profile table, we save `maximum_profile_table_rows_count` ids,
+  // each of which is generated by random. We use the FIFO strategy to eliminate the
+  // record when it reaches the upper limit. When we receive a record, we update the
+  // `cursor_` and get the corresponding id through it. Put operation will directly cover
+  // the previous data, so that we can avoid a delete operation.
+  Status status = gcs_table_storage_->ProfileTable().Put(ids_[cursor_++ % ids_.size()],
                                                          *profile_table_data, on_done);
   if (!status.ok()) {
     on_done(status);

src/ray/gcs/gcs_server/stats_handler_impl.h

@@ -14,6 +14,7 @@
 
 #pragma once
 
+#include "ray/common/ray_config.h"
 #include "ray/gcs/gcs_server/gcs_table_storage.h"
 #include "ray/gcs/redis_gcs_client.h"
 #include "ray/rpc/gcs_server/gcs_rpc_server.h"
@@ -25,7 +26,12 @@ namespace rpc {
 class DefaultStatsHandler : public rpc::StatsHandler {
  public:
   explicit DefaultStatsHandler(std::shared_ptr<gcs::GcsTableStorage> gcs_table_storage)
-      : gcs_table_storage_(std::move(gcs_table_storage)) {}
+      : gcs_table_storage_(std::move(gcs_table_storage)) {
+    for (int index = 0; index < RayConfig::instance().maximum_profile_table_rows_count();
+         ++index) {
+      ids_.emplace_back(UniqueID::FromRandom());
+    }
+  }
 
   void HandleAddProfileData(const AddProfileDataRequest &request,
                             AddProfileDataReply *reply,
@@ -36,6 +42,9 @@ class DefaultStatsHandler : public rpc::StatsHandler {
                                rpc::SendReplyCallback send_reply_callback) override;
 
  private:
+  int cursor_ = 0;
+  std::vector<UniqueID> ids_;
+
   std::shared_ptr<gcs::GcsTableStorage> gcs_table_storage_;
 };