diff --git a/src/ray/common/ray_config_def.h b/src/ray/common/ray_config_def.h index f7281ba04..f71d98a2e 100644 --- a/src/ray/common/ray_config_def.h +++ b/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) diff --git a/src/ray/gcs/gcs_client/test/global_state_accessor_test.cc b/src/ray/gcs/gcs_client/test/global_state_accessor_test.cc index 0b9d832f0..f86c9e187 100644 --- a/src/ray/gcs/gcs_client/test/global_state_accessor_test.cc +++ b/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) { diff --git a/src/ray/gcs/gcs_server/stats_handler_impl.cc b/src/ray/gcs/gcs_server/stats_handler_impl.cc index ae48b0bdd..6adce330b 100644 --- a/src/ray/gcs/gcs_server/stats_handler_impl.cc +++ b/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); diff --git a/src/ray/gcs/gcs_server/stats_handler_impl.h b/src/ray/gcs/gcs_server/stats_handler_impl.h index a73a717df..d9de7e40b 100644 --- a/src/ray/gcs/gcs_server/stats_handler_impl.h +++ b/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_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 ids_; + std::shared_ptr gcs_table_storage_; };