diff --git a/src/ray/common/ray_config_def.h b/src/ray/common/ray_config_def.h index ec47bd2db..0a43c32d1 100644 --- a/src/ray/common/ray_config_def.h +++ b/src/ray/common/ray_config_def.h @@ -234,6 +234,8 @@ RAY_CONFIG(uint32_t, gcs_create_actor_retry_interval_ms, 200) RAY_CONFIG(uint32_t, gcs_create_placement_group_retry_interval_ms, 200) /// Maximum number of destroyed actors in GCS server memory cache. RAY_CONFIG(uint32_t, maximum_gcs_destroyed_actor_cached_count, 10000) +/// Maximum number of dead nodes in GCS server memory cache. +RAY_CONFIG(uint32_t, maximum_gcs_dead_node_cached_count, 1000) /// Maximum number of times to retry putting an object when the plasma store is full. /// Can be set to -1 to enable unlimited retries. diff --git a/src/ray/gcs/gcs_client/test/service_based_gcs_client_test.cc b/src/ray/gcs/gcs_client/test/service_based_gcs_client_test.cc index 91e9aa08e..9969b75f4 100644 --- a/src/ray/gcs/gcs_client/test/service_based_gcs_client_test.cc +++ b/src/ray/gcs/gcs_client/test/service_based_gcs_client_test.cc @@ -29,7 +29,8 @@ class ServiceBasedGcsClientTest : public ::testing::Test { ServiceBasedGcsClientTest() { RayConfig::instance().initialize( {{"ping_gcs_rpc_server_max_retries", std::to_string(60)}, - {"maximum_gcs_destroyed_actor_cached_count", std::to_string(10)}}); + {"maximum_gcs_destroyed_actor_cached_count", std::to_string(10)}, + {"maximum_gcs_dead_node_cached_count", std::to_string(10)}}); TestSetupUtil::StartUpRedisServers(std::vector()); } @@ -571,6 +572,18 @@ class ServiceBasedGcsClientTest : public ::testing::Test { ASSERT_TRUE(actor.state() == expected_state); } + absl::flat_hash_set RegisterNodeAndMarkDead(int node_count) { + absl::flat_hash_set node_ids; + for (int index = 0; index < node_count; ++index) { + auto node_info = Mocker::GenNodeInfo(); + auto node_id = NodeID::FromBinary(node_info->node_id()); + EXPECT_TRUE(RegisterNode(*node_info)); + EXPECT_TRUE(UnregisterNode(node_id)); + node_ids.insert(node_id); + } + return node_ids; + } + // GCS server. gcs::GcsServerConfig config_; std::unique_ptr gcs_server_; @@ -1328,6 +1341,29 @@ TEST_F(ServiceBasedGcsClientTest, TestRandomEvictDestroyedActors) { EXPECT_TRUE(WaitForCondition(condition, timeout_ms_.count())); } +TEST_F(ServiceBasedGcsClientTest, TestEvictExpiredDeadNodes) { + // Simulate the scenario of node dead. + int node_count = RayConfig::instance().maximum_gcs_dead_node_cached_count(); + RegisterNodeAndMarkDead(node_count); + + // Restart GCS. + RestartGcsServer(); + + const auto &node_ids = RegisterNodeAndMarkDead(node_count); + + // Get all nodes. + auto condition = [this]() { + return GetNodeInfoList().size() == + RayConfig::instance().maximum_gcs_dead_node_cached_count(); + }; + EXPECT_TRUE(WaitForCondition(condition, timeout_ms_.count())); + + auto nodes = GetNodeInfoList(); + for (const auto &node : nodes) { + EXPECT_TRUE(node_ids.contains(NodeID::FromBinary(node.node_id()))); + } +} + // TODO(sang): Add tests after adding asyncAdd } // namespace ray diff --git a/src/ray/gcs/gcs_server/gcs_node_manager.cc b/src/ray/gcs/gcs_server/gcs_node_manager.cc index 82a448fd6..299ebe4fb 100644 --- a/src/ray/gcs/gcs_server/gcs_node_manager.cc +++ b/src/ray/gcs/gcs_server/gcs_node_manager.cc @@ -168,7 +168,8 @@ GcsNodeManager::GcsNodeManager(boost::asio::io_service &main_io_service, main_io_service_.post([this, node_id] { if (auto node = RemoveNode(node_id, /* is_intended = */ false)) { node->set_state(rpc::GcsNodeInfo::DEAD); - RAY_CHECK(dead_nodes_.emplace(node_id, node).second); + node->set_timestamp(current_sys_time_ms()); + AddDeadNodeToCache(node); auto on_done = [this, node_id, node](const Status &status) { auto on_done = [this, node_id, node](const Status &status) { RAY_CHECK_OK(gcs_pub_sub_->Publish( @@ -213,7 +214,8 @@ void GcsNodeManager::HandleUnregisterNode(const rpc::UnregisterNodeRequest &requ RAY_LOG(INFO) << "Unregistering node info, node id = " << node_id; if (auto node = RemoveNode(node_id, /* is_intended = */ true)) { node->set_state(rpc::GcsNodeInfo::DEAD); - RAY_CHECK(dead_nodes_.emplace(node_id, node).second); + node->set_timestamp(current_sys_time_ms()); + AddDeadNodeToCache(node); auto on_done = [this, node_id, node, reply, send_reply_callback](const Status &status) { @@ -347,7 +349,7 @@ void GcsNodeManager::HandleDeleteResources(const rpc::DeleteResourcesRequest &re void GcsNodeManager::HandleSetInternalConfig(const rpc::SetInternalConfigRequest &request, rpc::SetInternalConfigReply *reply, rpc::SendReplyCallback send_reply_callback) { - auto on_done = [reply, send_reply_callback, request](const Status status) { + auto on_done = [reply, send_reply_callback, request](const Status &status) { RAY_LOG(DEBUG) << "Set internal config: " << request.config().DebugString(); GCS_RPC_SEND_REPLY(send_reply_callback, reply, status); }; @@ -359,7 +361,7 @@ void GcsNodeManager::HandleGetInternalConfig(const rpc::GetInternalConfigRequest rpc::GetInternalConfigReply *reply, rpc::SendReplyCallback send_reply_callback) { auto get_system_config = [reply, send_reply_callback]( - ray::Status status, + const ray::Status &status, const boost::optional &config) { if (config.has_value()) { reply->mutable_config()->CopyFrom(config.get()); @@ -377,7 +379,7 @@ void GcsNodeManager::HandleGetAllAvailableResources( for (const auto &iter : GetClusterRealtimeResources()) { rpc::AvailableResources resource; resource.set_node_id(iter.first.Binary()); - for (auto res : iter.second->GetResourceAmountMap()) { + for (const auto &res : iter.second->GetResourceAmountMap()) { (*resource.mutable_resources_available())[res.first] = res.second.ToDouble(); } reply->add_resources_list()->CopyFrom(resource); @@ -464,8 +466,13 @@ void GcsNodeManager::LoadInitialData(const EmptyCallback &done) { AddNode(std::make_shared(item.second)); } else if (item.second.state() == rpc::GcsNodeInfo::DEAD) { dead_nodes_.emplace(item.first, std::make_shared(item.second)); + sorted_dead_node_list_.emplace_back(item.first, item.second.timestamp()); } } + sorted_dead_node_list_.sort([](const std::pair &left, + const std::pair &right) { + return left.second < right.second; + }); auto get_node_resource_callback = [this, done](const std::unordered_map &result) { @@ -513,5 +520,15 @@ void GcsNodeManager::UpdatePlacementGroupLoad( }); } +void GcsNodeManager::AddDeadNodeToCache(std::shared_ptr node) { + if (dead_nodes_.size() >= RayConfig::instance().maximum_gcs_dead_node_cached_count()) { + dead_nodes_.erase(sorted_dead_node_list_.begin()->first); + sorted_dead_node_list_.erase(sorted_dead_node_list_.begin()); + } + auto node_id = NodeID::FromBinary(node->node_id()); + dead_nodes_.emplace(node_id, node); + sorted_dead_node_list_.emplace_back(node_id, node->timestamp()); +} + } // namespace gcs } // namespace ray diff --git a/src/ray/gcs/gcs_server/gcs_node_manager.h b/src/ray/gcs/gcs_server/gcs_node_manager.h index f0c38eb18..65680ecab 100644 --- a/src/ray/gcs/gcs_server/gcs_node_manager.h +++ b/src/ray/gcs/gcs_server/gcs_node_manager.h @@ -246,6 +246,12 @@ class GcsNodeManager : public rpc::NodeInfoHandler { }; private: + /// Add the dead node to the cache. If the cache is full, the earliest dead node is + /// evicted. + /// + /// \param node The node which is dead. + void AddDeadNodeToCache(std::shared_ptr node); + /// The main event loop for node failure detector. boost::asio::io_service &main_io_service_; /// Detector to detect the failure of node. @@ -256,6 +262,9 @@ class GcsNodeManager : public rpc::NodeInfoHandler { absl::flat_hash_map> alive_nodes_; /// Dead nodes. absl::flat_hash_map> dead_nodes_; + /// The nodes are sorted according to the timestamp, and the oldest is at the head of + /// the list. + std::list> sorted_dead_node_list_; /// Cluster resources. absl::flat_hash_map cluster_resources_; /// Listeners which monitors the addition of nodes. diff --git a/src/ray/protobuf/gcs.proto b/src/ray/protobuf/gcs.proto index 59b0767ba..ad7aa56f5 100644 --- a/src/ray/protobuf/gcs.proto +++ b/src/ray/protobuf/gcs.proto @@ -259,6 +259,8 @@ message GcsNodeInfo { // The port at which the node will expose metrics to. int32 metrics_export_port = 9; + // Timestamp that the node is dead. + int64 timestamp = 10; } // Represents the demand for a particular resource shape.